import os import argparse import json import urllib.request import tarfile import shutil import pwd import subprocess from pathlib import Path try: import bcrypt BCRYPT_AVAILABLE = True except ImportError: BCRYPT_AVAILABLE = False SYSTEMD_SERVICE_PATH = "/etc/systemd/system/node_exporter.service" NODE_EXPORTER_BIN = "/usr/local/bin/node_exporter" NODE_EXPORTER_DIR = "/etc/node_exporter" def run_safe(cmd): subprocess.run(cmd, shell=isinstance(cmd, str), check=True) def fetch_latest_download_url(): url = "https://api.github.com/repos/prometheus/node_exporter/releases/latest" with urllib.request.urlopen(url) as response: data = json.loads(response.read().decode()) for asset in data["assets"]: if "linux-amd64" in asset["browser_download_url"]: return asset["browser_download_url"] raise RuntimeError("Nie znaleziono odpowiedniego URL do pobrania") def download_and_extract(url, extract_path="/tmp"): file_name = url.split("/")[-1] download_path = os.path.join(extract_path, file_name) print(f"Pobieranie: {url}") urllib.request.urlretrieve(url, download_path) with tarfile.open(download_path, "r:gz") as tar: tar.extractall(path=extract_path) for entry in os.listdir(extract_path): full_path = os.path.join(extract_path, entry) if entry.startswith("node_exporter") and os.path.isdir(full_path): return full_path raise RuntimeError("Nie znaleziono rozpakowanego katalogu node_exporter") def ensure_node_exporter_user(): try: pwd.getpwnam("node_exporter") except KeyError: run_safe(["useradd", "-rs", "/bin/false", "node_exporter"]) def install_binary(source_path, force_update=False): if os.path.exists(NODE_EXPORTER_BIN): if not force_update: print("Binarka już istnieje. Użyj --update aby zaktualizować.") return subprocess.run(["systemctl", "stop", "node_exporter"], check=False) shutil.copy(os.path.join(source_path, "node_exporter"), NODE_EXPORTER_BIN) os.chmod(NODE_EXPORTER_BIN, 0o755) print("Zainstalowano node_exporter.") def update_basic_auth_user(credential: str): if not BCRYPT_AVAILABLE: print("Błąd: brak modułu 'bcrypt'. Zainstaluj go poleceniem:\n pip install bcrypt") return if ":" not in credential: print("Błąd: użyj formatu --set-password user:haslo") return user, plain_password = credential.split(":", 1) config_path = os.path.join(NODE_EXPORTER_DIR, "config.yml") if not os.path.exists(config_path): print("Plik config.yml nie istnieje – uruchom instalację z --secured najpierw.") return hashed = bcrypt.hashpw(plain_password.encode(), bcrypt.gensalt()).decode() with open(config_path, "r") as f: lines = f.readlines() new_lines = [] in_auth = False user_updated = False for line in lines: if line.strip().startswith("basic_auth_users:"): in_auth = True new_lines.append(line) continue if in_auth and line.startswith(" ") and ":" in line: current_user = line.split(":")[0].strip() if current_user == user: new_lines.append(f" {user}: {hashed}\n") user_updated = True else: new_lines.append(line) elif in_auth and not line.startswith(" "): if not user_updated: new_lines.append(f" {user}: {hashed}\n") user_updated = True in_auth = False new_lines.append(line) else: new_lines.append(line) if in_auth and not user_updated: new_lines.append(f" {user}: {hashed}\n") with open(config_path, "w") as f: f.writelines(new_lines) print(f"Zmieniono hasło dla użytkownika '{user}'.") def write_systemd_service(secured=False): exec_line = f'{NODE_EXPORTER_BIN} --web.config.file="{NODE_EXPORTER_DIR}/config.yml"' if secured else NODE_EXPORTER_BIN content = f"""[Unit] Description=Node Exporter Wants=network-online.target After=network-online.target [Service] User=node_exporter ExecStart={exec_line} [Install] WantedBy=default.target """ with open(SYSTEMD_SERVICE_PATH, "w") as f: f.write(content) def setup_ssl_and_auth(): os.makedirs(NODE_EXPORTER_DIR, exist_ok=True) run_safe([ "openssl", "req", "-new", "-newkey", "rsa:4096", "-days", "3650", "-nodes", "-x509", "-subj", "/C=PL/ST=X/L=X/O=linuxiarz.pl/CN=*.linuxiarz.pl", "-keyout", f"{NODE_EXPORTER_DIR}/node_exporter.key", "-out", f"{NODE_EXPORTER_DIR}/node_exporter.crt" ]) config = """basic_auth_users: root: $2y$10$SNr5iyJMvqiecOx6tXgDTuBpxyd40Byp2j.iBM5lR/oQnlpi8nAje tls_server_config: cert_file: node_exporter.crt key_file: node_exporter.key """ with open(os.path.join(NODE_EXPORTER_DIR, "config.yml"), "w") as f: f.write(config) shutil.chown(NODE_EXPORTER_DIR, user="node_exporter", group="node_exporter") for file in os.listdir(NODE_EXPORTER_DIR): shutil.chown(os.path.join(NODE_EXPORTER_DIR, file), user="node_exporter", group="node_exporter") def enable_and_start_service(): run_safe(["systemctl", "daemon-reload"]) run_safe(["systemctl", "enable", "--now", "node_exporter"]) def uninstall(): subprocess.run(["systemctl", "stop", "node_exporter"], check=False) subprocess.run(["systemctl", "disable", "node_exporter"], check=False) if os.path.exists(SYSTEMD_SERVICE_PATH): os.remove(SYSTEMD_SERVICE_PATH) if os.path.exists(NODE_EXPORTER_BIN): os.remove(NODE_EXPORTER_BIN) if os.path.isdir(NODE_EXPORTER_DIR): shutil.rmtree(NODE_EXPORTER_DIR) run_safe(["systemctl", "daemon-reload"]) print("Node Exporter odinstalowany.") def main(): parser = argparse.ArgumentParser(description="Installer for Prometheus Node Exporter") parser.add_argument("--secured", action="store_true", help="Enable TLS and basic auth") parser.add_argument("--update", action="store_true", help="Force update of node_exporter binary") parser.add_argument("--uninstall", action="store_true", help="Uninstall node_exporter and clean files") parser.add_argument("--set-password", type=str, help="Ustaw basic auth (format: user:haslo), działa tylko z --secured") args = parser.parse_args() if args.uninstall: uninstall() return url = fetch_latest_download_url() extracted_dir = download_and_extract(url) ensure_node_exporter_user() install_binary(extracted_dir, force_update=args.update) if args.secured: setup_ssl_and_auth() write_systemd_service(secured=args.secured) enable_and_start_service() if args.set_password: update_basic_auth_user(args.set_password) print("Instalacja zakończona.") if __name__ == "__main__": main()