PaulVua/nebuleair_pro_4g
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 4 Wiki Activity

update

Browse Source
This commit is contained in:
Your Name
2025-02-03 13:27:21 +01:00
parent aaeb20aece
commit c6073b49b9
2 changed files with 29 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
SARA/SSL/full_test_HTTP.py
View File

@@ -43,20 +43,32 @@ config = load_config(config_file)
baudrate = config.get('SaraR4_baudrate', 115200)
send_uSpot = config.get('send_uSpot', False)
def read_complete_response(serial_connection, timeout=2, end_of_response_timeout=2):
def read_complete_response(serial_connection, timeout=2, end_of_response_timeout=2, wait_for_line=None):
response = bytearray()
serial_connection.timeout = timeout
end_time = time.time() + end_of_response_timeout
start_time = time.time()
while True:
elapsed_time = time.time() - start_time # Time since function start
if serial_connection.in_waiting > 0:
data = serial_connection.read(serial_connection.in_waiting)
response.extend(data)
end_time = time.time() + end_of_response_timeout # Reset timeout on new data
# Decode and check for the specific line
if wait_for_line:
decoded_response = response.decode('utf-8', errors='replace')
if wait_for_line in decoded_response:
print(f"[DEBUG] 🔎Found target line: {wait_for_line}")
break
elif time.time() > end_time:
print(f"[DEBUG] Timeout reached. No more data received.")
break
time.sleep(0.1) # Short sleep to prevent busy waiting
# Final response and debug output
total_elapsed_time = time.time() - start_time
print(f"[DEBUG] ⏱️ elapsed time: {total_elapsed_time:.2f}s. ⏱️")
return response.decode('utf-8', errors='replace')
ser_sara = serial.Serial(
@@ -97,7 +109,8 @@ try:
print("Trigger POST REQUEST")
command = f'AT+UHTTPC={profile_id},1,"/pro_4G/test.php","http.resp"\r'
ser_sara.write((command + '\r').encode('utf-8'))
response_SARA_6 = read_complete_response(ser_sara)
response_SARA_6 = read_complete_response(ser_sara, timeout=5, end_of_response_timeout=50, wait_for_line="+UUHTTPCR")
print(response_SARA_6)
time.sleep(1)

23
SARA/SSL/full_test_HTTPS_POST.py
View File

@@ -2,7 +2,7 @@
Script to set the URL for a HTTP request and trigger the POST Request
Ex:
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/SSL/full_test_HTTPS_POST.py ttyAMA2 api-prod.uspot.probesys.net /nebuleair?token=2AFF6dQk68daFZ
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/SSL/full_test_HTTPS_POST.py ttyAMA2 webhook.site /13502b8b-201a-41ea-ae33-983516074de5
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/SSL/full_test_HTTPS_POST.py ttyAMA2 webhook.site /0904d7b1-2558-43b9-8b35-df5bc40df967
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/SSL/full_test_HTTPS_POST.py ttyAMA2 aircarto.fr /tests/test.php
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/SSL/full_test_HTTPS_POST.py ttyAMA2 ssl.aircarto.fr /test.php
@@ -100,15 +100,15 @@ ser_sara = serial.Serial(
try:
#step 1: import the certificate
print("****")
with open("/var/www/nebuleair_pro_4g/SARA/SSL/certificate/e6.der", "rb") as cert_file:
certificate_name = "e6"
with open("/var/www/nebuleair_pro_4g/SARA/SSL/certificate/e6.pem", "rb") as cert_file:
certificate = cert_file.read()
size_of_string = len(certificate)
print("\033[0;33m Import certificate\033[0m")
# AT+USECMNG=0,<type>,<internal_name>,<data_size>
# type-> 0 -> trusted root CA
command = f'AT+USECMNG=0,0,"e6",{size_of_string}\r'
command = f'AT+USECMNG=0,0,"{certificate_name}",{size_of_string}\r'
ser_sara.write((command + '\r').encode('utf-8'))
response_SARA_1 = read_complete_response(ser_sara)
print(response_SARA_1)
@@ -149,15 +149,15 @@ try:
# op_code: 1 -> minimum SSL/TLS version
# param_val : 0 -> any; server can use any version for the connection; 1-> LSv1.0; 2->TLSv1.1; 3->TLSv1.2;
print("\033[0;33mSet the security profile (params)\033[0m")
minimum_SSL_version = 3
minimum_SSL_version = 0
command = f'AT+USECPRF={security_profile_id},1,{minimum_SSL_version}\r'
ser_sara.write((command + '\r').encode('utf-8'))
response_SARA_5bb = read_complete_response(ser_sara, wait_for_line="OK")
print(response_SARA_5bb)
time.sleep(0.5)
#op_code: 2 -> cipher suite
# 0 (factory-programmed value): (0x0000) Automatic the cipher suite will be negotiated in the handshake process
#op_code: 2 -> legacy cipher suite selection
# 0 (factory-programmed value): a list of default cipher suites is proposed at the beginning of handshake process, and a cipher suite will be negotiated among the cipher suites proposed in the list.
print("\033[0;33mSet cipher \033[0m")
cipher_suite = 0
command = f'AT+USECPRF={security_profile_id},2,{cipher_suite}\r'
@@ -168,7 +168,7 @@ try:
# op_code: 3 -> trusted root certificate internal name
print("\033[0;33mSet the security profile (choose cert)\033[0m")
command = f'AT+USECPRF={security_profile_id},3,"e6"\r'
command = f'AT+USECPRF={security_profile_id},3,"{certificate_name}"\r'
ser_sara.write((command + '\r').encode('utf-8'))
response_SARA_5c = read_complete_response(ser_sara, wait_for_line="OK")
print(response_SARA_5c)
@@ -290,7 +290,7 @@ try:
# Wait for the +UUHTTPCR response
print("Waiting for +UUHTTPCR response...")
response_SARA_3 = read_complete_response(ser_sara, timeout=5, end_of_response_timeout=30, wait_for_line="+UUHTTPCR")
response_SARA_3 = read_complete_response(ser_sara, timeout=5, end_of_response_timeout=50, wait_for_line="+UUHTTPCR")
print("\033[0;34m")
print(response_SARA_3)
@@ -326,7 +326,7 @@ try:
print(response_SARA_8)
# Get error code
print("\033[0;33mEmpty Memory\033[0m")
print("\033[0;33mGet error code\033[0m")
command = f'AT+UHTTPER={profile_id}\r'
ser_sara.write(command.encode('utf-8'))
response_SARA_9 = read_complete_response(ser_sara, wait_for_line="OK")
@@ -340,9 +340,10 @@ try:
3 HTTP Protocol error class
10 Wrong HTTP API USAGE
error_code (for error_class 3)
error_code (for error_class 3 or 10)
0 No error
11 Server connection error
22 PSD or CSD connection not established
73 Secure socket connect error
'''