diff --git a/manage_hetzner_servers.py b/manage_hetzner_servers.py index 0eb0042..1cd76ee 100644 --- a/manage_hetzner_servers.py +++ b/manage_hetzner_servers.py @@ -3,29 +3,40 @@ import configparser from hcloud import Client import argparse import pathlib +import os +import sys UPDATE_SCRIPT_PATH = pathlib.Path(__file__).parent / "update_local_config.py" CONFIG_PATH = pathlib.Path(__file__).parent / "config.ini" + def get_hetzner_client(): config = configparser.ConfigParser() config.read(CONFIG_PATH) api_token = config["hetzner"]["api_token"] return Client(token=api_token) + def list_hetzner_servers(client): return client.servers.get_all() + def delete_hetzner_server(client, server): server.delete() + def run_update_script(provider, location, server_type, interface, config_path): command = [ - "python", UPDATE_SCRIPT_PATH, - "--provider", provider, - "--location", location, - "--server_type", server_type, - "--interface", interface + os.path.realpath(sys.executable), + UPDATE_SCRIPT_PATH, + "--provider", + provider, + "--location", + location, + "--server_type", + server_type, + "--interface", + interface, ] if config_path: @@ -36,12 +47,19 @@ def run_update_script(provider, location, server_type, interface, config_path): raise Exception(f"Error executing update script: {result.stderr}") print(result.stdout) + def main(): - parser = argparse.ArgumentParser(description="Manage Hetzner servers and update local Wireguard VPN.") + parser = argparse.ArgumentParser( + description="Manage Hetzner servers and update local Wireguard VPN." + ) parser.add_argument("--location", type=str, required=True, help="Server location") parser.add_argument("--server_type", type=str, required=True, help="Server type") - parser.add_argument("--interface", type=str, required=True, help="Wireguard interface (e.g., wg0)") - parser.add_argument("--config_path", type=str, help="Path to the Wireguard configuration file") + parser.add_argument( + "--interface", type=str, required=True, help="Wireguard interface (e.g., wg0)" + ) + parser.add_argument( + "--config_path", type=str, help="Path to the Wireguard configuration file" + ) args = parser.parse_args() @@ -63,5 +81,6 @@ def main(): delete_hetzner_server(client, server) print(f"Deleted server: {server.name}") + if __name__ == "__main__": - main() \ No newline at end of file + main() diff --git a/update_local_config.py b/update_local_config.py index 2c30ca6..db7bf10 100644 --- a/update_local_config.py +++ b/update_local_config.py @@ -2,46 +2,70 @@ import subprocess import re import tempfile import os +import sys import argparse import pathlib SCRIPT_PATH = pathlib.Path(__file__).parent / "worker.py" + def execute_script(provider, location, server_type): command = [ - "python", SCRIPT_PATH, - "--provider", provider, - "--location", location, - "--server_type", server_type + os.path.realpath(sys.executable), + SCRIPT_PATH, + "--provider", + provider, + "--location", + location, + "--server_type", + server_type, ] result = subprocess.run(command, capture_output=True, text=True) if result.returncode != 0: raise Exception(f"Error executing script: {result.stderr}") return result.stdout.strip() + def replace_peer_section(config_content, new_peer_section): # Regular expression to match the [Peer] section - peer_section_pattern = re.compile(r'\[Peer\](?:\n(?!\[)[^\n]*)*') - + peer_section_pattern = re.compile(r"\[Peer\](?:\n(?!\[)[^\n]*)*") + # Find the existing [Peer] section match = peer_section_pattern.search(config_content) - + if match: # Replace the existing [Peer] section with the new one - updated_content = config_content[:match.start()] + new_peer_section + config_content[match.end():] + updated_content = ( + config_content[: match.start()] + + new_peer_section + + config_content[match.end() :] + ) else: # If no [Peer] section exists, append the new one updated_content = config_content + "\n" + new_peer_section - + return updated_content + def main(): - parser = argparse.ArgumentParser(description="Update local Wireguard configuration with a new peer.") - parser.add_argument("--provider", type=str, choices=["hetzner", "aws", "digitalocean", "azure"], required=True, help="Cloud provider") + parser = argparse.ArgumentParser( + description="Update local Wireguard configuration with a new peer." + ) + parser.add_argument( + "--provider", + type=str, + choices=["hetzner", "aws", "digitalocean", "azure"], + required=True, + help="Cloud provider", + ) parser.add_argument("--location", type=str, required=True, help="Server location") parser.add_argument("--server_type", type=str, required=True, help="Server type") - parser.add_argument("--interface", type=str, required=True, help="Wireguard interface (e.g., wg0)") - parser.add_argument("--config_path", type=str, help="Path to the Wireguard configuration file") + parser.add_argument( + "--interface", type=str, required=True, help="Wireguard interface (e.g., wg0)" + ) + parser.add_argument( + "--config_path", type=str, help="Path to the Wireguard configuration file" + ) args = parser.parse_args() @@ -55,7 +79,7 @@ def main(): new_peer_section = execute_script(provider, location, server_type) # Step 2: Read the local Wireguard configuration file - with open(config_path, 'r') as file: + with open(config_path, "r") as file: config_content = file.read() # Step 3: Replace the existing [Peer] section with the new one @@ -66,7 +90,7 @@ def main(): temp_file = pathlib.Path(tempdir) / f"{interface}.conf" temp_file_path = str(temp_file) - with open(temp_file_path, 'w') as file: + with open(temp_file_path, "w") as file: file.write(updated_content) # Step 5: Apply the updated Wireguard configuration @@ -76,7 +100,10 @@ def main(): # Overwrite the original config file with the updated content os.replace(temp_file_path, config_path) - print(f"Local Wireguard configuration for {interface} updated and applied successfully.") + print( + f"Local Wireguard configuration for {interface} updated and applied successfully." + ) + if __name__ == "__main__": - main() \ No newline at end of file + main()