#!/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 sys import json from requests_oauthlib import OAuth1Session def fail(message): print(json.dumps({"error": message})) sys.exit(1) def succeed(data): print(json.dumps(data)) sys.exit(0) class MaasManager: def __init__(self, config_path): self.config_path = config_path self.data = self.parse_json() self.session = self.init_session() def parse_json(self): try: with open(self.config_path, "r") as f: json_data = json.load(f) extension = json_data.get("externaldetails", {}).get("extension", {}) host = json_data.get("externaldetails", {}).get("host", {}) vm = json_data.get("externaldetails", {}).get("virtualmachine", {}) endpoint = host.get("endpoint") or extension.get("endpoint") apikey = host.get("apikey") or extension.get("apikey") distro_series = ( json_data.get("cloudstack.vm.details", {}) .get("details", {}) .get("distro_series", None) ) if not endpoint or not apikey: fail("Missing MAAS endpoint or apikey") if not endpoint.startswith("http://") and not endpoint.startswith("https://"): endpoint = "http://" + endpoint endpoint = endpoint.rstrip("/") parts = apikey.split(":") if len(parts) != 3: fail("Invalid apikey format. Expected consumer:token:secret") consumer, token, secret = parts system_id = ( json_data.get("cloudstack.vm.details", {}) .get("details", {}) .get("maas_system_id") or vm.get("maas_system_id", "") ) vm_name = vm.get("vm_name") or json_data.get("cloudstack.vm.details", {}).get("name") if not vm_name: vm_name = f"cs-{system_id}" if system_id else "cs-unknown" return { "endpoint": endpoint, "consumer": consumer, "token": token, "secret": secret, "distro_series": distro_series or "ubuntu/focal", "system_id": system_id, "vm_name": vm_name, } except Exception as e: fail(f"Error parsing JSON: {str(e)}") def init_session(self): return OAuth1Session( self.data["consumer"], resource_owner_key=self.data["token"], resource_owner_secret=self.data["secret"], ) def call_maas(self, method, path, data=None): if not path.startswith("/"): path = "/" + path url = f"{self.data['endpoint']}:5240/MAAS/api/2.0{path}" resp = self.session.request(method, url, data=data) if not resp.ok: fail(f"MAAS API error: {resp.status_code} {resp.text}") try: return resp.json() if resp.text else {} except ValueError: return {} def prepare(self): machines = self.call_maas("GET", "/machines/") ready = [m for m in machines if m.get("status_name") == "Ready"] if not ready: fail("No Ready machines available") sysid = self.data.get("system_id") if sysid: match = next((m for m in ready if m["system_id"] == sysid), None) if not match: fail(f"Provided system_id '{sysid}' not found among Ready machines") system = match else: system = ready[0] system_id = system["system_id"] mac = system.get("interface_set", [{}])[0].get("mac_address") hostname = system.get("hostname", "") if not mac: fail("No MAC address found") # Load original JSON so we can update nics with open(self.config_path, "r") as f: json_data = json.load(f) if json_data.get("cloudstack.vm.details", {}).get("nics"): json_data["cloudstack.vm.details"]["nics"][0]["mac"] = mac console_url = f"http://{self.data['endpoint'].replace('http://','').replace('https://','')}:5240/MAAS/r/machine/{system_id}/summary" result = { "nics": json_data["cloudstack.vm.details"]["nics"], "details": { "External:mac_address": mac, "External:maas_system_id": system_id, "External:hostname": hostname, "External:console_url": console_url, }, } succeed(result) def create(self): sysid = self.data.get("system_id") if not sysid: fail("system_id missing for create") ds = self.data.get("distro_series", "ubuntu/focal") vm_name = self.data.get("vm_name") # Cloud-init userdata to disable netplan, flush IPs on ens35, and run dhclient userdata = """#cloud-config network: config: disabled runcmd: - [ sh, -c, "dhclient -v -4 ens35 || true" ] """ self.call_maas( "POST", f"/machines/{sysid}/", { "op": "deploy", "distro_series": ds, "userdata": userdata, }, ) succeed({"status": "success", "message": f"Instance created with {self.data['distro_series']}"}) def delete(self): sysid = self.data.get("system_id") if not sysid: fail("system_id missing for delete") self.call_maas("POST", f"/machines/{sysid}/", {"op": "release"}) succeed({"status": "success", "message": "Instance deleted"}) def start(self): sysid = self.data.get("system_id") if not sysid: fail("system_id missing for start") self.call_maas("POST", f"/machines/{sysid}/", {"op": "power_on"}) succeed({"status": "success", "power_state": "PowerOn"}) def stop(self): sysid = self.data.get("system_id") if not sysid: fail("system_id missing for stop") self.call_maas("POST", f"/machines/{sysid}/", {"op": "power_off"}) succeed({"status": "success", "power_state": "PowerOff"}) def reboot(self): sysid = self.data.get("system_id") if not sysid: fail("system_id missing for reboot") self.call_maas("POST", f"/machines/{sysid}/", {"op": "power_cycle"}) succeed({"status": "success", "power_state": "PowerOn"}) def status(self): sysid = self.data.get("system_id") if not sysid: fail("system_id missing for status") resp = self.call_maas("GET", f"/machines/{sysid}/") state = resp.get("power_state", "") if state == "on": mapped = "PowerOn" elif state == "off": mapped = "PowerOff" else: mapped = "PowerUnknown" succeed({"status": "success", "power_state": mapped}) def main(): if len(sys.argv) < 3: fail("Usage: maas.py ") action = sys.argv[1].lower() json_file = sys.argv[2] try: manager = MaasManager(json_file) except FileNotFoundError: fail(f"JSON file not found: {json_file}") actions = { "prepare": manager.prepare, "create": manager.create, "delete": manager.delete, "start": manager.start, "stop": manager.stop, "reboot": manager.reboot, "status": manager.status, } if action not in actions: fail("Invalid action") actions[action]() if __name__ == "__main__": main()