qwphr/ess-moxa-configuration-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

CUBE configuration tool for INOX

Browse Source
This commit is contained in:
Quentin WEPHRE
2025-03-26 13:37:49 +01:00
parent 21c8c300c0
commit 62cb98f732
3 changed files with 189 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -11,3 +11,4 @@ lib
lib64
pyvenv.cfg
*.pem
*.csv

46
Python/cube_activate_ssh.py
View File

@@ -9,19 +9,17 @@ import io
load_dotenv(override=True)
def authenticate(base_url, username, password, certificate_path, verify_ssl=True):
def authenticate(base_url):
"""
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)
username = os.getenv("DEFAULT_CUBE_WEB_ADMIN_USER")
password = os.getenv("DEFAULT_CUBE_WEB_ADMIN_PASSWORD")
certificate = os.getenv("DEFAULT_CERTIFICATE").encode("utf-8")
# print(os.getenv("DEFAULT_CERTIFICATE").encode("utf-8"))
# Prepare the multipart form data
auth_params = {
"login": username,
@@ -29,13 +27,12 @@ def authenticate(base_url, username, password, certificate_path, verify_ssl=True
}
files = {
"params": (None, json.dumps(auth_params), "application/json"),
"certificate": ("certificate.pem", os.getenv("DEFAULT_CERTIFICATE").encode("utf-8"), "application/octet-stream")
"certificate": ("certificate.pem", certificate, "application/octet-stream")
}
# print(files)
try:
print(f"Authenticating as {username}...")
response = requests.post(auth_url, files=files, verify=verify_ssl)
response = requests.post(auth_url, files=files, verify=False)
response.raise_for_status() # Raise exception for 4XX/5XX responses
# Extract token from response
@@ -44,7 +41,6 @@ def authenticate(base_url, username, password, certificate_path, verify_ssl=True
if not token:
print("Error: No token received in authentication response")
sys.exit(1)
print("Authentication successful.")
return token
@@ -53,9 +49,9 @@ def authenticate(base_url, username, password, certificate_path, verify_ssl=True
print(f"Authentication failed: {e}")
if hasattr(e, 'response') and e.response:
print(f"Response: {e.response.text}")
sys.exit(1)
raise
def set_ssh_status(base_url, token, verify_ssl=True):
def set_ssh_status(base_url, token):
"""
Set SSH status (enable) using the provided JWT token.
"""
@@ -71,7 +67,7 @@ def set_ssh_status(base_url, token, verify_ssl=True):
try:
print(f"Sending request to enable SSH...")
response = requests.post(ssh_url, headers=headers, json=payload, verify=verify_ssl)
response = requests.post(ssh_url, headers=headers, json=payload, verify=False)
response.raise_for_status()
print(f"SSH enabled successfully!")
@@ -84,21 +80,10 @@ def set_ssh_status(base_url, token, verify_ssl=True):
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()
def activate_ssh(ip_address):
# Ensure the URL uses HTTPS
url = args.url
url = ip_address
if not url.startswith("https://"):
# Convert http:// to https:// or add https:// if no protocol specified
if url.startswith("http://"):
@@ -107,16 +92,15 @@ def main():
else:
url = "https://" + url
print(f"Adding HTTPS protocol: {url}")
if not url.endswith(":9080"):
url = url + ":9080"
verify_ssl = False
if not verify_ssl:
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
token = authenticate(url, args.username, args.password, args.certificate, verify_ssl)
token = authenticate(url)
if not token:
return
set_ssh_status(url, token, verify_ssl)
if __name__ == "__main__":
main()
set_ssh_status(url, token)

169
Python/cube_ssh_batch.py Normal file
View File

@@ -0,0 +1,169 @@
import csv
import paramiko
import time
from cube_activate_ssh import activate_ssh
from dotenv import load_dotenv
import os
import re
from azure.iot.hub import IoTHubRegistryManager
from azure.iot.hub.models import Twin, TwinProperties
load_dotenv(override=True)
ip_address_prefix = "10.188.11."
ssh_command = "hostname"
csv_filename = "hoohana6.csv"
SITE_NAME = "HOOHANA"
ssh_username = os.getenv("DEFAULT_CUBE_LINUX_ADMIN_USER")
ssh_password = os.getenv("DEFAULT_CUBE_LINUX_ADMIN_PASSWORD")
CONNECTION_STRING = str(os.getenv("CONNECTION_STRING_INOX_PROD"))
def execute_ssh_command(ip, command):
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
client.connect(ip, port=11022, username=ssh_username, password=ssh_password, allow_agent=False, look_for_keys=False)
stdin, stdout, stderr = client.exec_command(command)
result = stdout.read().decode().lower().strip()
return result
except Exception as e:
print(f"SSH Error: {str(e)}")
raise
finally:
client.close()
def update_cloud_config(ip, new_content):
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
client.connect(ip, port=11022, username=ssh_username, password=ssh_password, allow_agent=False, look_for_keys=False)
stdin, stdout, stderr = client.exec_command(f'sudo -S bash -c \'cat > /etc/cube/config-azure.properties << EOF\n{new_content}\nEOF\'\n')
stdin.write(ssh_password + "\n")
stdin.flush()
stdoutput = [line for line in stdout]
stderroutput = [line for line in stderr]
for output in stdoutput:
print(output.strip())
except Exception as e:
print(f"SSH Error: {str(e)}")
raise
finally:
client.close()
def main():
print("Starting...")
with open(csv_filename, mode="w", newline="") as file:
writer = csv.writer(file)
writer.writerow(["Number", "IP address", "Cube ID", "Environment", "Correct configuration"])
registry_manager = IoTHubRegistryManager.from_connection_string(CONNECTION_STRING)
results = []
for i in range (54, 55):
ip_address = f"{ip_address_prefix}{i}"
print(f"Activating SSH for {ip_address}:", end=" ")
try:
activate_ssh(ip_address)
except Exception as e:
print("Failed!")
writer.writerow([i, ip_address, "UNREACHABLE", "NA", "NA"])
file.flush()
continue
print("Activated!")
print(f"Executing {ssh_command} for {ip_address}:", end=" ")
try:
cube_id = execute_ssh_command(ip_address, ssh_command)
except Exception as e:
print("Failed!")
writer.writerow([i, ip_address, "UNREACHABLE", "NA", "NA"])
file.flush()
continue
print(cube_id)
print(f"Getting configured Connection String")
try:
connection_string = execute_ssh_command(ip_address, "grep \"connection-string\" /etc/cube/config-azure.properties")
if connection_string == "":
raise Exception("No Connection String extracted!")
iothub_match = re.search(r"hostname\\=(.*?);", connection_string, re.IGNORECASE)
iothub = iothub_match.group(1) if iothub_match else None
if iothub.lower() == "IotHub-CUBE-PROD.azure-devices.net".lower():
migration = "SAFT"
elif iothub.lower() == "iot-ingest-ess-prod.azure-devices.net".lower():
migration = "INOX"
else:
migration = "NONE"
device_id_match = re.search(r"deviceid\\=(.*?);", connection_string, re.IGNORECASE)
cloud_cube_id = device_id_match.group(1) if device_id_match else None
if cloud_cube_id.lower() == cube_id.lower():
status = "CORRECT"
else:
status = "INCORRECT"
except Exception as e:
print(e)
migration = "NONE"
status = "INCORRECT"
if migration == "SAFT" or migration == "NONE":
print("SAFT device, migrating to INOX...")
print("Creating device on INOX...")
try:
registry_manager.create_device_with_sas(
cube_id,
primary_key="", secondary_key="",
status="enabled",
iot_edge=False
)
except Exception as iot_e:
print("Error creating new device!")
print(iot_e)
continue
print("Adding tags to new device...")
try:
twin = registry_manager.get_twin(cube_id)
twin_patch = Twin(properties=TwinProperties(desired={}), tags={
"site": SITE_NAME,
"number": i
})
registry_manager.update_twin(cube_id, twin_patch, twin.etag)
except Exception as iot_e:
print("Error assigning tags to new device!")
print(iot_e)
print("Requesting primary key...")
try:
device_info = registry_manager.get_device(cube_id)
primary_key = device_info.authentication.symmetric_key.primary_key
new_connection_string = f"HostName\\={CONNECTION_STRING.split(';')[0].split('=')[1]};DeviceId\\={cube_id};SharedAccessKey\\={primary_key}"
new_content = f'light-telemetry=false\ncompression-enabled=true\ntelemetry-on=true\nremote-update-on=true\nconnection-string={new_connection_string}'
except Exception as iot_e:
print("Error getting new Connection String!")
print(iot_e)
continue
print("Setting new Connection String...")
try:
update_cloud_config(ip_address, new_content)
except Exception as ssh_e:
print("Error when setting the new Connection String!")
print(ssh_e)
continue
print("Done!")
writer.writerow([i, ip_address, cube_id, migration, status])
file.flush()
if __name__ == "__main__":
main()