From 21c8c300c0ed64c5d60f452e52ee9f1ffb5c7ca8 Mon Sep 17 00:00:00 2001 From: Quentin WEPHRE Date: Tue, 25 Mar 2025 11:27:04 +0100 Subject: [PATCH] Added default arguments for CUBE SSH activation. --- .gitignore | 1 + Moulinette/requirements.txt | 2 + Python/azure_iot_hub_list_devices.py | 44 ++++------ Python/cube_activate_ssh.py | 122 +++++++++++++++++++++++++++ Python/isight_device.py | 23 +++++ Python/requirements.txt | 2 + 6 files changed, 168 insertions(+), 26 deletions(-) create mode 100644 Python/cube_activate_ssh.py create mode 100644 Python/isight_device.py create mode 100644 Python/requirements.txt diff --git a/.gitignore b/.gitignore index 6ac9c0e..78e4c87 100644 --- a/.gitignore +++ b/.gitignore @@ -10,3 +10,4 @@ bin lib lib64 pyvenv.cfg +*.pem \ No newline at end of file diff --git a/Moulinette/requirements.txt b/Moulinette/requirements.txt index b0da061..cf52f81 100644 --- a/Moulinette/requirements.txt +++ b/Moulinette/requirements.txt @@ -1,2 +1,4 @@ pandas==1.4.3 openpyxl==3.1.0 +requests +dotenv-python diff --git a/Python/azure_iot_hub_list_devices.py b/Python/azure_iot_hub_list_devices.py index fc97a1a..db5fc84 100644 --- a/Python/azure_iot_hub_list_devices.py +++ b/Python/azure_iot_hub_list_devices.py @@ -1,7 +1,8 @@ from azure.iot.hub import IoTHubRegistryManager -from azure.iot.hub.protocol.models import QuerySpecification +from azure.iot.hub.protocol.models import QuerySpecification, Module from azure.iot.hub.models import CloudToDeviceMethod, CloudToDeviceMethodResult from dotenv import load_dotenv +from isight_device import iSightDevice import json import os @@ -24,36 +25,27 @@ if CONNECTION_STRING == "": registry_manager = IoTHubRegistryManager.from_connection_string(CONNECTION_STRING) -query_spec = QuerySpecification(query="SELECT * FROM devices WHERE IS_DEFINED(tags.site) AND capabilities.iotEdge = true AND tags.site = 'PIERREFONDS'") +query_spec = QuerySpecification(query="SELECT * FROM devices WHERE IS_DEFINED(tags.site) AND capabilities.iotEdge = true ") query_result = registry_manager.query_iot_hub(query_spec) devices = [] for item in query_result.items: - deviceId = str(item.device_id) - site = str(item.tags['site']) - number = int(item.tags['number']) - cloud_version = str(item.tags['version']) - devices.append([deviceId, site, number, cloud_version]) + # currentDevice = iSightDevice(str(item.device_id), str(item.tags['site']), int(item.tags['number']), str(item.tags['version'])) + devices.append(iSightDevice(str(item.device_id), str(item.tags['site']), int(item.tags['number']), str(item.tags['version']))) -ordered_devices = sorted(devices, key = lambda x: (x[1], x[2])) +devices.sort(key = lambda d: (d.site, d.number)) -for index in range(len(ordered_devices)): - current_device_modules = registry_manager.get_modules(ordered_devices[index][0]) +for device in devices: + print(device, end="\t") + current_device_modules = registry_manager.get_modules(device.deviceId) for module in current_device_modules: - if module.module_id == module_id: - thingspro_module = module - print(thingspro_module) - try: - direct_method = CloudToDeviceMethod(method_name=method_name, payload=json.loads(payload)) - response = registry_manager.invoke_device_module_method(device_id=ordered_devices[index][0], module_id=module_id, direct_method_request=direct_method) - # print(str(ordered_devices[index][0]), str(ordered_devices[index][1]), str(ordered_devices[index][2]), response.payload['data']['description'], "device version: " + str(response.payload['data']['firmwareVersion']), "cloud version: " + str(ordered_devices[index][3]), sep=";") - device_version = str(response.payload['data']['firmwareVersion']) - if device_version != "1.6.0": - payload = '{"deleteFileAfterInstallComplete": true, "install": true, "url": "https://files.thingsprocloud.com/package/Upgrade_AIG-301_2.5.0-4404_IMG_1.5_to_1.6.0.yaml"}' - direct_method = CloudToDeviceMethod(method_name=method_name, payload=json.loads(payload)) - except: - print(str(ordered_devices[index][0]), str(ordered_devices[index][1]), str(ordered_devices[index][2]), "UNREACHABLE", "device version: UNREACHABLE", "cloud version: " + str(ordered_devices[index][3]), sep=";") - else: - print("No thingspro-agent available for " + ordered_devices[index][0] + " (" + ordered_devices[index][1] + " " + ordered_devices[index][2] + ")") - \ No newline at end of file + if (module.module_id == "thingspro-agent"): + device.setModule(module) + try: + direct_method = CloudToDeviceMethod(method_name=method_name, payload=json.loads(payload)) + response = registry_manager.invoke_device_module_method(device_id=device.deviceId, module_id=module_id, direct_method_request=direct_method) + device_version = str(response.payload['data']['firmwareVersion']) + print(device_version) + except: + print("ERROR") \ No newline at end of file diff --git a/Python/cube_activate_ssh.py b/Python/cube_activate_ssh.py new file mode 100644 index 0000000..e49ce3a --- /dev/null +++ b/Python/cube_activate_ssh.py @@ -0,0 +1,122 @@ +import requests +import json +import argparse +import sys +import os +import urllib3 +from dotenv import load_dotenv +import io + +load_dotenv(override=True) + +def authenticate(base_url, username, password, certificate_path, verify_ssl=True): + """ + Authenticate with the CUBE API using username, password and certificate. + Returns the JWT token if successful. + """ + auth_url = f"{base_url}/api/auth" + + # Verify certificate file exists + # if not os.path.isfile(certificate_path): + # print(f"Error: Certificate file not found at: {certificate_path}") + # sys.exit(1) + + # print(os.getenv("DEFAULT_CERTIFICATE").encode("utf-8")) + # Prepare the multipart form data + auth_params = { + "login": username, + "password": password + } + files = { + "params": (None, json.dumps(auth_params), "application/json"), + "certificate": ("certificate.pem", os.getenv("DEFAULT_CERTIFICATE").encode("utf-8"), "application/octet-stream") + } + # print(files) + + try: + print(f"Authenticating as {username}...") + response = requests.post(auth_url, files=files, verify=verify_ssl) + response.raise_for_status() # Raise exception for 4XX/5XX responses + + # Extract token from response + auth_data = response.json() + token = auth_data.get("token") + + if not token: + print("Error: No token received in authentication response") + sys.exit(1) + + print("Authentication successful.") + return token + + except requests.exceptions.RequestException as e: + print(f"Authentication failed: {e}") + if hasattr(e, 'response') and e.response: + print(f"Response: {e.response.text}") + sys.exit(1) + +def set_ssh_status(base_url, token, verify_ssl=True): + """ + Set SSH status (enable) using the provided JWT token. + """ + ssh_url = f"{base_url}/api/ssh" + + headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {token}" + } + + # Set new SSH status + payload = { "currentStatus": True } + + try: + print(f"Sending request to enable SSH...") + response = requests.post(ssh_url, headers=headers, json=payload, verify=verify_ssl) + response.raise_for_status() + + print(f"SSH enabled successfully!") + + return True + + except requests.exceptions.RequestException as e: + print(f"SSH activation failed: {e}") + if hasattr(e, 'response') and e.response: + print(f"Response: {e.response.text}") + return False + +def main(): + parser = argparse.ArgumentParser(description="Manage SSH on CUBE application") + parser.add_argument("--url", help="Base URL of the CUBE API (e.g., https://cube-04fe12:9080)", + default="https://cube-04fe12:9080") + parser.add_argument("--username", help="Admin username with ROLE_SAFT_ADMIN permissions", + default=os.getenv("DEFAULT_CUBE_WEB_ADMIN_USER")) + parser.add_argument("--password", help="Admin password", + default=os.getenv("DEFAULT_CUBE_WEB_ADMIN_PASSWORD")) + parser.add_argument("--certificate", help="Path to mission certificate file", + default=os.getenv("DEFAULT_CERTIFICATE")) + + args = parser.parse_args() + + # Ensure the URL uses HTTPS + url = args.url + if not url.startswith("https://"): + # Convert http:// to https:// or add https:// if no protocol specified + if url.startswith("http://"): + url = "https://" + url[7:] + print(f"Converting to HTTPS: {url}") + else: + url = "https://" + url + print(f"Adding HTTPS protocol: {url}") + + verify_ssl = False + if not verify_ssl: + urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) + + token = authenticate(url, args.username, args.password, args.certificate, verify_ssl) + if not token: + return + + set_ssh_status(url, token, verify_ssl) + +if __name__ == "__main__": + main() diff --git a/Python/isight_device.py b/Python/isight_device.py new file mode 100644 index 0000000..264d8f6 --- /dev/null +++ b/Python/isight_device.py @@ -0,0 +1,23 @@ +from azure.iot.hub.protocol.models import Module +from typing import Optional + +class iSightDevice: + def __init__(self, deviceId: str, site: str, number: int, cloudVersion: str): + if not isinstance(deviceId, str) or not isinstance(site, str) or not isinstance(cloudVersion, str): + raise TypeError("deviceId, site, and cloudVersion must be strings.") + if not isinstance(number, int) or number < 0: + raise ValueError("number must be a non-negative integer.") + + self.deviceId = deviceId + self.site = site + self.number = number + self.cloudVersion = cloudVersion + self.thingsproModule: Optional[Module] = None + + def setModule(self, module: Module): + if not isinstance(module, Module): + raise TypeError("module must be an instance of azure.iot.hub.protocol.models.module_py3.Module") + self.thingsproModule = module + + def __str__(self): + return f"{self.deviceId} {self.site} {self.number} {self.cloudVersion}" \ No newline at end of file diff --git a/Python/requirements.txt b/Python/requirements.txt new file mode 100644 index 0000000..839ba48 --- /dev/null +++ b/Python/requirements.txt @@ -0,0 +1,2 @@ +requests +dotenv-python \ No newline at end of file