| #!/usr/bin/env python3 |
| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance |
| # with the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, |
| # software distributed under the License is distributed on an |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| # KIND, either express or implied. See the License for the |
| # specific language governing permissions and limitations |
| # under the License. |
| |
| import warnings |
| warnings.filterwarnings('ignore') |
| |
| import json |
| import sys |
| import winrm |
| |
| |
| def fail(message): |
| print(json.dumps({"error": message})) |
| sys.exit(1) |
| |
| |
| def succeed(data): |
| print(json.dumps(data)) |
| sys.exit(0) |
| |
| |
| class HyperVManager: |
| def __init__(self, config_path): |
| self.config_path = config_path |
| self.data = self.parse_json() |
| self.session = self.init_winrm_session() |
| |
| def parse_json(self): |
| try: |
| with open(self.config_path, 'r') as f: |
| json_data = json.load(f) |
| |
| external_host_details = json_data["externaldetails"].get("host", []) |
| data = { |
| "url": external_host_details["url"], |
| "username": external_host_details["username"], |
| "password": external_host_details["password"], |
| "network_switch": external_host_details["network_switch"], |
| "vhd_path": external_host_details["vhd_path"], |
| "vm_path": external_host_details["vm_path"], |
| "cert_validation": "validate" if external_host_details.get("verify_tls_certificate", "true").lower() == "true" else "ignore" |
| } |
| |
| external_vm_details = json_data["externaldetails"].get("virtualmachine", []) |
| if external_vm_details: |
| data["template_type"] = external_vm_details["template_type"] |
| data["generation"] = external_vm_details.get("generation", 1) |
| data["template_path"] = external_vm_details.get("template_path", "") |
| data["iso_path"] = external_vm_details.get("iso_path", "") |
| data["vhd_size_gb"] = external_vm_details.get("vhd_size_gb", "") |
| |
| data["cpus"] = json_data["cloudstack.vm.details"]["cpus"] |
| data["memory"] = json_data["cloudstack.vm.details"]["minRam"] |
| data["vmname"] = json_data["cloudstack.vm.details"]["name"] |
| |
| nics = json_data["cloudstack.vm.details"].get("nics", []) |
| data["nics"] = [] |
| for nic in nics: |
| data["nics"].append({ |
| "mac": nic["mac"], |
| "vlan": nic["broadcastUri"].replace("vlan://", "") |
| }) |
| |
| parameters = json_data.get("parameters", []) |
| if parameters: |
| data["snapshot_name"] = parameters.get("snapshot_name", "") |
| |
| return data |
| |
| except KeyError as e: |
| fail(f"Missing required field in JSON: {str(e)}") |
| except Exception as e: |
| fail(f"Error parsing JSON: {str(e)}") |
| |
| def init_winrm_session(self): |
| return winrm.Session( |
| f"https://{self.data['url']}:5986/wsman", |
| auth=(self.data["username"], self.data["password"]), |
| transport='ntlm', |
| server_cert_validation=self.data["cert_validation"] |
| ) |
| |
| def run_ps_int(self, command): |
| r = self.session.run_ps(command) |
| if r.status_code != 0: |
| raise Exception(r.std_err.decode()) |
| return r.std_out.decode() |
| |
| def run_ps(self, command): |
| try: |
| output = self.run_ps_int(command) |
| return output |
| except Exception as e: |
| fail(str(e)) |
| |
| def vm_not_present(self, exception): |
| vm_not_present_str = f'Hyper-V was unable to find a virtual machine with name "{self.data["vmname"]}"' |
| return vm_not_present_str in str(exception) |
| |
| def create(self): |
| vm_name = self.data["vmname"] |
| cpus = self.data["cpus"] |
| memory = self.data["memory"] |
| memory_mb = int(memory) / 1024 / 1024 |
| template_path = self.data["template_path"] |
| vhd_path = self.data["vhd_path"] + "\\" + vm_name + ".vhdx" |
| vhd_size_gb = self.data["vhd_size_gb"] |
| generation = self.data["generation"] |
| iso_path = self.data["iso_path"] |
| network_switch = self.data["network_switch"] |
| vm_path = self.data["vm_path"] |
| template_type = self.data.get("template_type", "template") |
| |
| vhd_created = False |
| vm_created = False |
| vm_started = False |
| try: |
| command = ( |
| f'New-VM -Name "{vm_name}" -MemoryStartupBytes {memory_mb}MB ' |
| f'-Generation {generation} -Path "{vm_path}" ' |
| ) |
| if template_type == "iso": |
| if (iso_path == ""): |
| fail("Missing required field in JSON: iso_path") |
| if (vhd_size_gb == ""): |
| fail("Missing required field in JSON: vhd_size_gb") |
| command += ( |
| f'-NewVHDPath "{vhd_path}" -NewVHDSizeBytes {vhd_size_gb}GB; ' |
| f'Add-VMDvdDrive -VMName "{vm_name}" -Path "{iso_path}"; ' |
| ) |
| else: |
| if (template_path == ""): |
| fail("Missing required field in JSON: template_path") |
| self.run_ps_int(f'Copy-Item "{template_path}" "{vhd_path}"') |
| vhd_created = True |
| command += f'-VHDPath "{vhd_path}"; ' |
| |
| self.run_ps_int(command) |
| vm_created = True |
| |
| command = f'Remove-VMNetworkAdapter -VMName "{vm_name}" -Name "Network Adapter" -ErrorAction SilentlyContinue; ' |
| self.run_ps_int(command) |
| |
| command = f'Set-VMProcessor -VMName "{vm_name}" -Count "{cpus}"; ' |
| if (generation == 2): |
| command += f'Set-VMFirmware -VMName "{vm_name}" -EnableSecureBoot Off; ' |
| |
| self.run_ps_int(command) |
| |
| for idx, nic in enumerate(self.data["nics"]): |
| adapter_name = f"NIC{idx+1}" |
| self.run_ps_int(f'Add-VMNetworkAdapter -VMName "{vm_name}" -SwitchName "{network_switch}" -Name "{adapter_name}"') |
| self.run_ps_int(f'Set-VMNetworkAdapter -VMName "{vm_name}" -Name "{adapter_name}" -StaticMacAddress "{nic["mac"]}"') |
| self.run_ps_int(f'Set-VMNetworkAdapterVlan -VMName "{vm_name}" -VMNetworkAdapterName "{adapter_name}" -Access -VlanId "{nic["vlan"]}"') |
| |
| self.run_ps_int(f'Start-VM -Name "{vm_name}"') |
| vm_started = True |
| |
| succeed({"status": "success", "message": "Instance created"}) |
| |
| except Exception as e: |
| if vm_started: |
| self.run_ps_int(f'Stop-VM -Name "{vm_name}" -Force -TurnOff') |
| if vm_created: |
| self.run_ps_int(f'Remove-VM -Name "{vm_name}" -Force') |
| if vhd_created: |
| self.run_ps_int(f'Remove-Item -Path "{vhd_path}" -Force') |
| fail(str(e)) |
| |
| def start(self): |
| self.run_ps(f'Start-VM -Name "{self.data["vmname"]}"') |
| succeed({"status": "success", "message": "Instance started"}) |
| |
| def stop(self): |
| try: |
| self.run_ps_int(f'Stop-VM -Name "{self.data["vmname"]}" -Force') |
| except Exception as e: |
| if self.vm_not_present(e): |
| succeed({"status": "success", "message": "Instance stopped"}) |
| else: |
| fail(str(e)) |
| succeed({"status": "success", "message": "Instance stopped"}) |
| |
| def reboot(self): |
| self.run_ps(f'Restart-VM -Name "{self.data["vmname"]}" -Force') |
| succeed({"status": "success", "message": "Instance rebooted"}) |
| |
| def status(self): |
| command = f'(Get-VM -Name "{self.data["vmname"]}").State' |
| state = self.run_ps(command) |
| power_state = "unknown" |
| if state.strip().lower() == "running": |
| power_state = "poweron" |
| elif state.strip().lower() == "off": |
| power_state = "poweroff" |
| succeed({"status": "success", "power_state": power_state}) |
| |
| def delete(self): |
| try: |
| self.run_ps_int(f'Remove-VM -Name "{self.data["vmname"]}" -Force') |
| except Exception as e: |
| if self.vm_not_present(e): |
| succeed({"status": "success", "message": "Instance deleted"}) |
| else: |
| fail(str(e)) |
| succeed({"status": "success", "message": "Instance deleted"}) |
| |
| def suspend(self): |
| self.run_ps(f'Suspend-VM -Name "{self.data["vmname"]}"') |
| succeed({"status": "success", "message": "Instance suspended"}) |
| |
| def resume(self): |
| self.run_ps(f'Resume-VM -Name "{self.data["vmname"]}"') |
| succeed({"status": "success", "message": "Instance resumed"}) |
| |
| def create_snapshot(self): |
| snapshot_name = self.data["snapshot_name"] |
| if snapshot_name == "": |
| fail("Missing required field in JSON: snapshot_name") |
| command = f'Checkpoint-VM -VMName "{self.data["vmname"]}" -SnapshotName "{snapshot_name}"' |
| self.run_ps(command) |
| succeed({"status": "success", "message": f"Snapshot '{snapshot_name}' created"}) |
| |
| def list_snapshots(self): |
| command = ( |
| f'Get-VMSnapshot -VMName "{self.data["vmname"]}" ' |
| '| Select-Object Name, @{Name="CreationTime";Expression={$_.CreationTime.ToString("s")}} ' |
| '| ConvertTo-Json' |
| ) |
| snapshots = json.loads(self.run_ps(command)) |
| succeed({"status": "success", "printmessage": "true", "message": snapshots}) |
| |
| def restore_snapshot(self): |
| snapshot_name = self.data["snapshot_name"] |
| if snapshot_name == "": |
| fail("Missing required field in JSON: snapshot_name") |
| command = f'Restore-VMSnapshot -VMName "{self.data["vmname"]}" -Name "{snapshot_name}" -Confirm:$false' |
| self.run_ps(command) |
| succeed({"status": "success", "message": f"Snapshot '{snapshot_name}' restored"}) |
| |
| def delete_snapshot(self): |
| snapshot_name = self.data["snapshot_name"] |
| if snapshot_name == "": |
| fail("Missing required field in JSON: snapshot_name") |
| command = f'Remove-VMSnapshot -VMName "{self.data["vmname"]}" -Name "{snapshot_name}" -Confirm:$false' |
| self.run_ps(command) |
| succeed({"status": "success", "message": f"Snapshot '{snapshot_name}' deleted"}) |
| |
| |
| def main(): |
| if len(sys.argv) < 3: |
| fail("Usage: script.py <operation> '<json-file-path>'") |
| |
| operation = sys.argv[1].lower() |
| json_file_path = sys.argv[2] |
| |
| try: |
| manager = HyperVManager(json_file_path) |
| except FileNotFoundError: |
| fail(f"JSON file not found: {json_file_path}") |
| except json.JSONDecodeError: |
| fail("Invalid JSON in file") |
| |
| operations = { |
| "create": manager.create, |
| "start": manager.start, |
| "stop": manager.stop, |
| "reboot": manager.reboot, |
| "delete": manager.delete, |
| "status": manager.status, |
| "suspend": manager.suspend, |
| "resume": manager.resume, |
| "listsnapshots": manager.list_snapshots, |
| "createsnapshot": manager.create_snapshot, |
| "restoresnapshot": manager.restore_snapshot, |
| "deletesnapshot": manager.delete_snapshot |
| } |
| |
| if operation not in operations: |
| fail("Invalid action") |
| |
| try: |
| operations[operation]() |
| except Exception as e: |
| fail(str(e)) |
| |
| |
| if __name__ == "__main__": |
| main() |
