407 lines
15 KiB
Python
407 lines
15 KiB
Python
#!/usr/bin/python
|
|
# -*- coding: utf-8 -*-
|
|
|
|
# Copyright: (c) 2023, Florian L. <git+ansible@pounce.tech>
|
|
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
|
|
from __future__ import annotations
|
|
__metaclass__ = type
|
|
|
|
DOCUMENTATION = r'''
|
|
---
|
|
module: github_artifact
|
|
|
|
short_description: fetch assets from a GitHub repository release
|
|
|
|
description: vaguely similar to a package manager, but for GitHub artifacts.
|
|
|
|
version_added: "2.15.0"
|
|
|
|
options:
|
|
asset_name:
|
|
description: filename of the asset to retrieve, used only for release type; supports templating
|
|
type: str
|
|
required: false
|
|
default: ""
|
|
asset_type:
|
|
description: whether the asset is a release or just a tagged asset
|
|
type: str
|
|
required: true
|
|
choices:
|
|
- release
|
|
- tag
|
|
cmds:
|
|
description: commands to execute in order to install the downloaded asset; supports templating
|
|
type: list
|
|
elements: str
|
|
required: false
|
|
default: []
|
|
creates:
|
|
description: if provided and target file / directory exists, this step will **not** be run unless `--force` is specified
|
|
type: str
|
|
required: false
|
|
force:
|
|
description: forces the re-installation of the package no matter its state
|
|
type: bool
|
|
required: false
|
|
default: false
|
|
github_token:
|
|
description: a GitHub app token if you have one; limits impact of rate-limiting errors
|
|
type: str
|
|
required: false
|
|
repository:
|
|
description: repository to query, formatted like "<owner>/<repo>"
|
|
required: true
|
|
type: str
|
|
version:
|
|
description: version of the asset to fetch; defaults to `latest`
|
|
required: false
|
|
type: str
|
|
default: latest
|
|
|
|
notes:
|
|
- "Strings that allow the use of templating variables support the following:"
|
|
- " V(version): version of the system, B(NOT) the asset's;"
|
|
- " V(system): type of the OS, retrieved from C(platform.system), e.g. I(Linux), I(Darwin);"
|
|
- " V(machine): machine architecture, retrieved from C(platform.machine), e.g. I(x86_64), I(i386);"
|
|
- " V(asset_name): name of the selected asset from the GitHub metadata results, e.g. I(dive_0.11.0_linux_amd64.deb);"
|
|
- " V(asset_dirname): directory where the downloaded asset is located, e.g. I(/tmp/ansible-moduletmp-1695757626.5862153-xjpc5ip8);"
|
|
- " V(asset_filename): name of the download asset file, e.g. I(dive_0.11.0_linux_amd64bnha_1dr.deb);"
|
|
- " V(asset_version): version of the asset, retrieved directly from the GitHub metadata;"
|
|
- " all variables defined in C(/etc/os-release), lowercase."
|
|
|
|
author:
|
|
- Florian L. (@NaeiKinDus)
|
|
'''
|
|
|
|
EXAMPLES = r'''
|
|
- name: Install dependencies from GitHub
|
|
become: true
|
|
tags:
|
|
- molecule-idempotence-notest
|
|
nullified.infrastructure.github_artifact:
|
|
asset_type: tag
|
|
repository: smxi/inxi
|
|
cmds:
|
|
- tar -zxf {asset_dirname}/{asset_filename}
|
|
- install --group=root --mode=755 --owner=root smxi-inxi-*/inxi /usr/bin
|
|
- install --group=root --mode=644 --owner=root smxi-inxi-*/inxi.1 /usr/share/man/man1
|
|
- apt-get install libdata-dump-perl
|
|
creates: /usr/bin/inxi
|
|
'''
|
|
|
|
RETURN = r'''
|
|
# These are examples of possible return values, and in general should use other names for return values.
|
|
original_message:
|
|
description: The original name param that was passed in.
|
|
type: str
|
|
returned: always
|
|
sample: "hello world"
|
|
'''
|
|
|
|
from ansible.module_utils.basic import AnsibleModule, missing_required_lib # noqa
|
|
from ansible.module_utils.urls import fetch_url, fetch_file # noqa
|
|
|
|
ANSIBLE_MODULE: AnsibleModule | None = None
|
|
LIB_IMPORTED = False
|
|
GITHUB_API_BASE = "https://api.github.com"
|
|
GITHUB_DOWNLOAD_BASE = "https://github.com"
|
|
GITHUB_API_VERSION = "2022-11-28"
|
|
DEFAULT_HEADERS: dict[str, str] = {
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json, application/vnd.github+json;q=0.8",
|
|
}
|
|
|
|
TEMPLATE_ASSET_NAME_VARS: dict[str, str] = {
|
|
"version": "", # platform.version(), e.g. "12 (bookworm)"
|
|
"system": "", # platform.system(), e.g. "Linux", "Darwin"
|
|
"machine": "" # platform.machine(), e.g. "x86_64", "i386"
|
|
}
|
|
try:
|
|
from datetime import datetime
|
|
from difflib import SequenceMatcher
|
|
from json import loads
|
|
from os import environ, sep, path
|
|
from platform import system, machine
|
|
from typing import Any
|
|
LIB_IMPORTED = True
|
|
except ModuleNotFoundError as excp:
|
|
import traceback
|
|
IMPORT_LIB_ERROR = traceback.format_exc()
|
|
IMPORT_LIB_NAME = excp.name
|
|
try:
|
|
from platform import freedesktop_os_release
|
|
except (ModuleNotFoundError, ImportError):
|
|
FREEDESKTOP_OS_RELEASE_FILE = '/etc/os-release'
|
|
|
|
def freedesktop_os_release() -> dict[str, str]:
|
|
try:
|
|
with open(FREEDESKTOP_OS_RELEASE_FILE, 'r') as fd:
|
|
data = fd.read()
|
|
os_info: dict[str, str] = {key: value.strip('"') for key, value in (line.split('=') for line in data.splitlines())}
|
|
except FileNotFoundError:
|
|
return dict()
|
|
return os_info
|
|
|
|
|
|
def find_compatible_asset(assets_list: list[dict[str, str | int | float]], asset_name: str) -> dict[str, str] | None:
|
|
""" takes a list of assets and tries to find the most relevant one; assumes only one asset is required """
|
|
best_match: dict[str, str] = {}
|
|
matched_name_ratio: float = 0.0
|
|
|
|
if len(assets_list) == 0:
|
|
return None
|
|
elif len(assets_list) == 1:
|
|
return {
|
|
"asset_name": assets_list[0]["name"],
|
|
"download_url": assets_list[0]["browser_download_url"],
|
|
"match_ratio": "1.0"
|
|
}
|
|
|
|
sm = SequenceMatcher(a=asset_name, b="")
|
|
for asset in assets_list:
|
|
if asset_name == asset["name"]:
|
|
return {
|
|
"asset_name": asset["name"],
|
|
"download_url": asset["browser_download_url"],
|
|
"match_ratio": "1.0"
|
|
}
|
|
sm.set_seq2(asset["name"])
|
|
ratio = sm.ratio()
|
|
if ratio > matched_name_ratio:
|
|
best_match = asset
|
|
matched_name_ratio = ratio
|
|
if not best_match:
|
|
return None
|
|
return {
|
|
"asset_name": best_match["name"],
|
|
"download_url": best_match["browser_download_url"],
|
|
"match_ratio": "{:.5f}".format(matched_name_ratio)
|
|
}
|
|
|
|
|
|
def fetch_github_data(url: str) -> tuple[dict | None, dict[str, int]]:
|
|
""" query GitHub API and return a JSON formatted response along with HTTP info data """
|
|
response, info = fetch_url(ANSIBLE_MODULE, url, headers=DEFAULT_HEADERS)
|
|
http_status: int = info.get("status", 999)
|
|
if http_status >= 400:
|
|
return None, info
|
|
return loads(response.read().decode("utf-8")), info
|
|
|
|
|
|
def get_released_asset(artifact: dict[str, str]) -> tuple[dict[str, str], dict[str, int] | None]:
|
|
""" fetch asset metadata using the release GitHub API """
|
|
repository: str = artifact["repository"]
|
|
version: str = artifact["version"]
|
|
releases_url: str = "{}/repos/{}/releases/{}{}".format(
|
|
GITHUB_API_BASE,
|
|
repository,
|
|
"tags/" if version != "latest" else "",
|
|
version
|
|
)
|
|
|
|
if ANSIBLE_MODULE.check_mode:
|
|
return {
|
|
"asset_name": "{}/{}.ext".format(repository, version),
|
|
"download_url": "download_url",
|
|
"match_confidence": "match_ratio",
|
|
"version": version
|
|
}, {}
|
|
|
|
response_data, info = fetch_github_data(releases_url)
|
|
if not response_data:
|
|
return {"error": "No release found for version {}. Requested source: {}".format(version, releases_url)}, info
|
|
|
|
asset_name = artifact.get("asset_name", "").format(**TEMPLATE_ASSET_NAME_VARS)
|
|
asset = find_compatible_asset(response_data["assets"], asset_name=asset_name)
|
|
if not asset:
|
|
if not asset_name:
|
|
return {"error": "No matching asset detected, try specifying the desired asset name in arguments list"}, info
|
|
return {"error": "No asset matching name {} found".format(asset_name)}, info
|
|
return {
|
|
"asset_name": asset["asset_name"],
|
|
"download_url": asset["download_url"],
|
|
"match_confidence": asset["match_ratio"],
|
|
"version": response_data["tag_name"] or response_data["name"]
|
|
}, info
|
|
|
|
|
|
def get_tagged_asset(artifact: dict[str, Any]) -> tuple[dict[str, str], dict[str, int] | None]:
|
|
""" fetch asset metadata using the tags GitHub API """
|
|
repository: str = artifact["repository"]
|
|
version: str = artifact["version"]
|
|
tags_url: str = "{}/repos/{}/tags?per_page=1".format(GITHUB_API_BASE, repository)
|
|
|
|
if version != "latest":
|
|
return {
|
|
"asset_name": "{}.tar.gz".format(version),
|
|
"download_url": "{}/{}/archive/refs/tags/{}.tar.gz".format(GITHUB_DOWNLOAD_BASE, repository, version),
|
|
"version": version
|
|
}, None
|
|
|
|
if ANSIBLE_MODULE.check_mode:
|
|
return {
|
|
"asset_name": "asset_name",
|
|
"download_url": "download_url",
|
|
"version": version
|
|
}, {}
|
|
|
|
response_data, info = fetch_github_data(tags_url)
|
|
if not response_data:
|
|
return {
|
|
"error": "No tagged asset found for '{}'".format(tags_url)
|
|
}, info
|
|
response_data = response_data[0]
|
|
return {
|
|
"asset_name": "{}.tar.gz".format(response_data.get("name", "unknown")),
|
|
"download_url": response_data.get("tarball_url"),
|
|
"version": response_data.get("name", "latest")
|
|
}, info
|
|
|
|
|
|
def fetch_metadata(artifact: dict[str, str | list[str]]) -> dict[str, str] | None:
|
|
""" retrieve metadata from the specified repository """
|
|
if artifact["asset_type"] == "tag":
|
|
metadata, info = get_tagged_asset(artifact)
|
|
else:
|
|
metadata, info = get_released_asset(artifact)
|
|
if info:
|
|
reset_date = info.get("x-ratelimit-reset", None)
|
|
metadata["rate_limit_max"] = info.get("x-ratelimit-limit", "unknown")
|
|
metadata["rate_limit_remaining"] = info.get("x-ratelimit-remaining", "unknown")
|
|
metadata["rate_limit_reset_date"] = datetime.fromtimestamp(float(reset_date)).isoformat() if reset_date else "unknown"
|
|
return metadata
|
|
|
|
|
|
def main():
|
|
global ANSIBLE_MODULE
|
|
|
|
module_args: dict[str, dict[str, Any]] = {
|
|
"asset_name": {
|
|
"type": "str",
|
|
"required": False,
|
|
"default": ""
|
|
},
|
|
"asset_type": {
|
|
"type": "str",
|
|
"required": True,
|
|
"choices": ["release", "tag"],
|
|
},
|
|
"cmds": {
|
|
"type": "list",
|
|
"elements": "str",
|
|
"required": False,
|
|
"default": []
|
|
},
|
|
"creates": {
|
|
"type": "str",
|
|
"required": False
|
|
},
|
|
"force": {
|
|
"type": "bool",
|
|
"required": False,
|
|
"default": False
|
|
},
|
|
"repository": {
|
|
"type": "str",
|
|
"required": True
|
|
},
|
|
"version": {
|
|
"type": "str",
|
|
"required": False,
|
|
"default": "latest"
|
|
},
|
|
"github_token": {
|
|
"type": "str",
|
|
"required": False,
|
|
"no_log": True
|
|
},
|
|
}
|
|
result: dict[str, Any] = {
|
|
"changed": False,
|
|
"commands": [],
|
|
"failed": False,
|
|
"filepath": "",
|
|
"msg": "",
|
|
"state": "",
|
|
"version": ""
|
|
}
|
|
|
|
ANSIBLE_MODULE = AnsibleModule(argument_spec=module_args, supports_check_mode=True)
|
|
if "FORCE_CHECK_MODE" in environ:
|
|
ANSIBLE_MODULE.check_mode = True if environ.get("FORCE_CHECK_MODE", False) in [True, "1", "True", "true"] else False
|
|
if ANSIBLE_MODULE.params["github_token"]:
|
|
DEFAULT_HEADERS["Authorization"] = "Bearer {}".format(ANSIBLE_MODULE.params["github_token"])
|
|
|
|
creates_file: str | None = ANSIBLE_MODULE.params.get("creates", None)
|
|
if creates_file and path.exists(creates_file) and not ANSIBLE_MODULE.params.get("force", False):
|
|
result["state"] = "ignored"
|
|
ANSIBLE_MODULE.exit_json(**result)
|
|
|
|
if not LIB_IMPORTED:
|
|
ANSIBLE_MODULE.fail_json(msg=missing_required_lib(IMPORT_LIB_NAME), exception=IMPORT_LIB_ERROR) # pylint: disable=used-before-assignment
|
|
|
|
# load local metadata cached file to retrieve installed version
|
|
TEMPLATE_ASSET_NAME_VARS.update({key.lower(): value for key, value in freedesktop_os_release().items()})
|
|
TEMPLATE_ASSET_NAME_VARS["system"] = system().lower()
|
|
TEMPLATE_ASSET_NAME_VARS["machine"] = machine().lower()
|
|
|
|
artifact: dict[str, str | list[str]] = {}
|
|
for param_name in ["asset_name", "asset_type", "cmds", "repository", "version"]:
|
|
artifact[param_name] = ANSIBLE_MODULE.params[param_name]
|
|
if not artifact["version"]:
|
|
artifact["version"] = "latest"
|
|
if not artifact["cmds"]:
|
|
artifact["cmds"] = []
|
|
|
|
asset_data: dict[str, str] = fetch_metadata(artifact)
|
|
result["rate_limit_remaining"] = asset_data.get("rate_limit_remaining", "unknown")
|
|
result["rate_limit_max"] = asset_data.get("rate_limit_max", "unknown")
|
|
result["version"] = asset_data.get("version", artifact.get("version"))
|
|
|
|
if "error" in asset_data:
|
|
result["state"] = "fetch failed"
|
|
result["msg"] = asset_data.get("error", "unknown error encountered")
|
|
result["failed"] = True
|
|
ANSIBLE_MODULE.fail_json(**result)
|
|
|
|
# download artifact
|
|
if ANSIBLE_MODULE.check_mode:
|
|
result["filepath"] = "unknown"
|
|
else:
|
|
result["filepath"] = fetch_file(ANSIBLE_MODULE, asset_data.get("download_url", "unknown"), decompress=False)
|
|
TEMPLATE_ASSET_NAME_VARS["asset_name"] = asset_data.get("asset_name", "unknown")
|
|
TEMPLATE_ASSET_NAME_VARS["asset_version"] = asset_data.get("version", "unknown")
|
|
parts = result["filepath"].rsplit(sep, 1)
|
|
TEMPLATE_ASSET_NAME_VARS["asset_dirname"] = parts[0] if len(parts) > 1 else ""
|
|
TEMPLATE_ASSET_NAME_VARS["asset_filename"] = parts[1] if len(parts) > 1 else parts[0]
|
|
|
|
# install artifact
|
|
artifact_commands = [line.format(**TEMPLATE_ASSET_NAME_VARS) for line in artifact["cmds"]]
|
|
if ANSIBLE_MODULE.check_mode:
|
|
result["commands"] = artifact_commands
|
|
result["state"] = "should be installed" if len(artifact_commands) else "should be downloaded"
|
|
else:
|
|
for command_line in artifact_commands:
|
|
cmd_rc, cmd_out, cmd_err = ANSIBLE_MODULE.run_command(command_line, use_unsafe_shell=True, cwd=ANSIBLE_MODULE.tmpdir)
|
|
result["changed"] = True
|
|
result["commands"].append({
|
|
"command": command_line,
|
|
"stdout": cmd_out,
|
|
"stderr": cmd_err,
|
|
"ret_code": cmd_rc
|
|
})
|
|
|
|
if cmd_rc:
|
|
result["state"] = "installation failed"
|
|
result["msg"] = cmd_err
|
|
result["failed"] = True
|
|
ANSIBLE_MODULE.fail_json(**result)
|
|
|
|
result["state"] = "installed" if len(artifact_commands) else "downloaded"
|
|
|
|
result["msg"] = "Successful"
|
|
ANSIBLE_MODULE.exit_json(**result)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|