IoT & Embedded Devices: Certificate Underpinning

Hi to everyone:

I’m having problems with this lab.

To begin with, it seems to me that the difficulty level of the tasks is much higher than the knowledge provided in the theoretical section.

Additionally, port 443 requires elevated privileges, so you can’t simply run python exploit.py, nor sudo python exploit.py. The only option I’ve found is running sudo /usr/bin/python3 exploit.py.

And finally—this may be my own mistake—I can’t manage to connect to the alarm. I’m attaching my script in case I could get some advice.

Thank you very much.

import socket
import ssl
import json
import threading

# Configuración
UDP_PORT = 8080
TCP_PORT = 443
HOST = '0.0.0.0'

def http_json_response(obj, status_code=200, reason="OK", keep_alive=True):
    body = json.dumps(obj).encode("utf-8")  # bytes
    headers = [
        f"HTTP/1.1 {status_code} {reason}",
        "Content-Type: application/json; charset=utf-8",
        f"Content-Length: {len(body)}",
        f"Connection: {'keep-alive' if keep_alive else 'close'}",
        "", 
        ""
    ]
    head = "\r\n".join(headers).encode("utf-8")
    return head + body
    
def handle_udp():
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as udp_sock:
            udp_sock.bind((HOST, UDP_PORT))
            print(f"[*] [UDP] Escuchando updates en el puerto {UDP_PORT}...")
            while True:
                data, addr = udp_sock.recvfrom(4096)
                print(f"[+] [UDP] Update de {addr}: {data.decode()}")
    except Exception as e:
        print(f"[!] [UDP] Error: {e}")

def start_tcp_tls_server():
    context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    context.load_cert_chain(certfile="cert.pem", keyfile="key.pem")

    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as tcp_sock:
        tcp_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        tcp_sock.bind((HOST, TCP_PORT))
        tcp_sock.listen(5)
        print(f"[*] [TCP] Servidor TLS listo en el puerto {TCP_PORT}...")

        while True:
            newsocket, addr = tcp_sock.accept()
            print(f"[+] [TCP] Conexión entrante de {addr}")
            try:
                with context.wrap_socket(newsocket, server_side=True) as tls_conn:
                    data = tls_conn.recv(4096).decode()
                    print(f"[>] [TCP] Recibido: {data}")

                    disarm_payload = {"ClientId": "AXG1337VFXL", "Status": "Disabled", "TS": "1486495687"}
                    
                    print(f"[*] [TCP] Enviando comando de desarmado...")
                    resp = http_json_response(disarm_payload)
                    tls_conn.sendall(resp)

                    result = tls_conn.recv(4096).decode()
                    print(f"\n[!] RESULTADO: {result}\n")

            except Exception as e:
                print(f"[!] [TCP] Error en handshake/envío: {e}")

if __name__ == "__main__":
    # 1. Hilo para udp listener
    udp_thread = threading.Thread(target=handle_udp, daemon=True)
    
    # 2. Inicio el upd listener
    udp_thread.start()
    
    # 3. Inicio el server
    start_tcp_tls_server()