import pandas as pd import requests from urllib3.exceptions import InsecureRequestWarning import jq import json import os from dotenv import load_dotenv load_dotenv() # Function to authenticate and get token def authenticate(device_ip, payload): auth_url = f"https://{device_ip}:8443/api/v1/auth" response = requests.post(auth_url, json=payload, verify=False) if response.status_code == 200: token = response.json()["data"]["token"] #print(f"Authentication successful. Token received: {token}") #print(" authenticated!") return token else: print(f"Authentication failed. Status code: {response.status_code}") return None # Function to send PATCH request def send_patch_request(device_ip, token, connection_string): headers = { "mx-api-token": token } payload = { "provisioning": { "source": "manual", "connectionString": connection_string, "enable": True } } patch_url = f"https://{device_ip}:8443/api/v1/azure-iotedge" response = requests.patch(patch_url, json=payload, headers=headers, verify=False) if response.status_code == 200: print(f"PATCH request successful for device {device_ip}") else: print(f"Failed to send PATCH request to device {device_ip}. Status code: {response.status_code}") # Function to send PATCH request def patch_time(device_ip, token): headers = { "mx-api-token": token } # payload = { # "ntp": { # "source": "timeserver", # "server": "10.84.171.254", # "enable": True # } # } payload = { "timezone": "America/Chicago" } patch_url = f"https://{device_ip}:8443/api/v1/device/time" response = requests.patch(patch_url, json=payload, headers=headers, verify=False) if response.status_code == 200: json_data = json.loads(response.content.decode()) time = json_data['data']['time'] timezone = json_data['data']['timezone'] last = json_data['data']['lastUpdateTime'] server = json_data['data']['ntp']['server'] enabled = json_data['data']['ntp']['enable'] print(time + " " + timezone + " " + last + " " + server + " " + str(enabled)) else: json_data = json.loads(response.content.decode()) #print(json.dumps(json_data, indent=2)) # Function to send UPGRADE request def send_upgrade_request(device_ip, token, upgrade_url): headers = { "mx-api-token": token } payload = { "download": True, "install": True, "url": upgrade_url, } patch_url = f"https://{device_ip}:8443/api/v1/upgrades" response = requests.post(patch_url, json=payload, headers=headers, verify=False) json_data = json.loads(response.content.decode()) id = json_data['data']['id'] return id # Function to send UPGRADE request def get_upgrade_job(device_ip, token, id): headers = { "mx-api-token": token } payload = { } patch_url = f"https://{device_ip}:8443/api/v1/upgrades/{id}" response = requests.get(patch_url, json=payload, headers=headers, verify=False) if response.status_code == 200: #print(f"GET request successful for device {device_ip}") #print(json_str) #print(json_data['data'][json_data['count'] - 1]['parameter']['url'], json_data['data'][json_data['count'] - 1]['state'], json_data['count']) json_data = json.loads(response.content.decode()) getid = json_data['data']['id'] created = json_data['data']['createdAt'] cur_status = json_data['data']['state'] current_tasks = json_data['data']['completedTask'] total_tasks = json_data['data']['totalTask'] print("JOB #" + str(getid) + " " + str(current_tasks) + "/" + str(total_tasks)) print("CREATED ON: " + str(created)) print("CURRENT STATUS: " + str(cur_status)) print(json.dumps(json_data, indent=2)) else: print(f"Failed to send GET request to device {device_ip}. Status code: {response.status_code}") print(response.content) def get_upgrade_jobs(device_ip, token): headers = { "mx-api-token": token } payload = { } patch_url = f"https://{device_ip}:8443/api/v1/upgrades" response = requests.get(patch_url, json=payload, headers=headers, verify=False) if response.status_code == 200: #print(f"GET request successful for device {device_ip}") #print(json_str) #print(json_data['data'][json_data['count'] - 1]['parameter']['url'], json_data['data'][json_data['count'] - 1]['state'], json_data['count']) json_data = json.loads(response.content.decode()) count = json_data['count'] print(str(count)) for i in range(count): getid = json_data['data'][i]['id'] created = json_data['data'][i]['createdAt'] cur_status = json_data['data'][i]['state'] current_tasks = json_data['data'][i]['completedTask'] total_tasks = json_data['data'][i]['totalTask'] print("JOB #" + str(getid) + " " + str(current_tasks) + "/" + str(total_tasks)) print("CREATED ON: " + str(created)) print("CURRENT STATUS: " + str(cur_status)) else: print(f"Failed to send GET request to device {device_ip}. Status code: {response.status_code}") print(response.content) # Function to send UPGRADE request def start_upgrade_job(device_ip, token, id): headers = { "mx-api-token": token } payload = { } patch_url = f"https://{device_ip}:8443/api/v1/upgrades/{id}/start" response = requests.put(patch_url, json=payload, headers=headers, verify=False) if response.status_code == 200: json_data = json.loads(response.content.decode()) curid = json_data['data']['id'] startedat = json_data['data']['startedAt'] print("Job #" + str(curid) + " started on " + str(startedat)) else: print(f"Failed to send GET request to device {device_ip}. Status code: {response.status_code}") print(response.content.decode()) # Function to send UPGRADE request def put_API(device_ip, token): headers = { "mx-api-token": token } payload = { } patch_url = f"https://{device_ip}:8443/api/v1/upgrades/3/start" response = requests.put(patch_url, json=payload, headers=headers, verify=False) if response.status_code == 200: #print(f"GET request successful for device {device_ip}") #json_data = json.loads(response.content.decode()) #json_str = json.dumps(json_data) #print(jq.compile('.data.completedAt').input(json.loads(json_str)).first()) json_data = json.loads(response.content.decode()) print(json_data['data']['firmwareVersion']) else: print(f"Failed to send GET request to device {device_ip}. Status code: {response.status_code}") print(response.content.decode()) def get_API(device_ip, token): headers = { "mx-api-token": token } payload = { } patch_url = f"https://{device_ip}:8443/api/v1/device/general" response = requests.get(patch_url, json=payload, headers=headers, verify=False) if response.status_code == 200: #print(f"GET request successful for device {device_ip}") #json_data = json.loads(response.content.decode()) #json_str = json.dumps(json_data) #print(jq.compile('.data.completedAt').input(json.loads(json_str)).first()) json_data = json.loads(response.content.decode()) hostname = json_data['data']['hostName'] serial = json_data['data']['serialNumber'] version = json_data['data']['firmwareVersion'] description = json_data['data']['description'] print(hostname + " " + serial + " " + version + " " + description) #json_str = json.dumps(json_data, indent=2) #print(json_str) else: print(f"Failed to send GET request to device {device_ip}. Status code: {response.status_code}") print(response.content.decode()) def get_time(device_ip, token): headers = { "mx-api-token": token } payload = { } patch_url = f"https://{device_ip}:8443/api/v1/device/time" response = requests.get(patch_url, json=payload, headers=headers, verify=False) if response.status_code == 200: #print(f"GET request successful for device {device_ip}") #json_data = json.loads(response.content.decode()) #json_str = json.dumps(json_data) #print(jq.compile('.data.completedAt').input(json.loads(json_str)).first()) json_data = json.loads(response.content.decode()) time = json_data['data']['time'] timezone = json_data['data']['timezone'] last = json_data['data']['lastUpdateTime'] print(time + " " + timezone + " " + last) #json_str = json.dumps(json_data, indent=2) #print(json_str) else: print(f"Failed to send GET request to device {device_ip}. Status code: {response.status_code}") print(response.content.decode()) def delete_API(device_ip, token): headers = { "mx-api-token": token } payload = { } patch_url = f"https://{device_ip}:8443/api/v1/upgrades/5" response = requests.delete(patch_url, json=payload, headers=headers, verify=False) print(response.status_code) json_data = json.loads(response.content.decode()) json_str = json.dumps(json_data, indent=2) print(json_str) # Read the Excel file # excel_file_path = "" # if excel_file_path == "": # print("Provide Excel file path before running the script!") # exit(11) # df = pd.read_excel(excel_file_path) # df = df[df["device_name"].notnull()] # Iterate over each row in the DataFrame requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning) # for index, row in df.iterrows(): # device_name = row['device_name'] # device_ip_address_https = row['device_ip_address_http'] # connection_string = row['connection_string'] # upgrade_url = "" #https://files.thingsprocloud.com/package/moxa-aig-301-series-includes-security-patch-firmware-v1.0.deb.yaml # if upgrade_url == "": # print("Provide upgrade URL before running the script!") # exit(12) # # Authenticate and get token # payload_auth = { # "acceptEULA": True, # "name": "", # "password": "" # } # if payload_auth["name"] == "" or payload_auth["password"] == "": # print("Provide the credentials before running the script!") # exit(10) # print(device_name, end="") # token = authenticate(device_ip_address_https, payload_auth) # if token: # get_API(device_ip_address_https, token) # print("\n") default_user = str(os.getenv("DEFAULT_MOXA_USER")) default_password = str(os.getenv("DEFAULT_MOXA_PASSWORD")) for i in range(193, 222): upgrade_url = "https://files.thingsprocloud.com/package/Upgrade_AIG-301_2.4.0-4020_IMG_1.4_to_1.5.deb.yaml" payload_auth = { "acceptEULA": True, "name": default_user, "password": default_password } device_ip_address = str("10.84.171." + str(i)) #print(device_ip_address, end="") token = authenticate(device_ip_address, payload_auth) if token: #id = send_upgrade_request(device_ip_address,token,upgrade_url) #print(id) #get_upgrade_job(device_ip_address, token, 6) #start_upgrade_job(device_ip_address, token, id) #put_API(device_ip_address, token) #patch_time(device_ip_address,token) #get_time(device_ip_address, token) #delete_API(device_ip_address,token) get_API(device_ip_address, token) else: print("Authentication failed!")