pyp100 module funktionstauglich

backend/handshake.py

@@ -0,0 +1,128 @@
+import requests
+import json
+
+# Constants from the Wireshark capture
+PUBLIC_KEY = """-----BEGIN PUBLIC KEY-----
+MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQCMl89OZsjqE8yZ9TQhUb9h539WTX3U8Y5YCNdp
+OhuXvLFYcAT5mvC074VFROmD0xhvw5hrwESOisqpPPU9r78JpLuYUKd+/aidvykqBT8OW5rDLb6d
+O9FO6Gc+bV8L8ttHVlBFoX69EqiRhcreGPG6FQz4JqGJF4T1nFi0EvALXwIDAQAB
+-----END PUBLIC KEY-----"""
+
+# Vorbereitete verschlüsselte Befehle (aus Wireshark extrahiert)
+COMMAND_ON = """ps0Puxc37EK4PhfcevceL3lyyDrjwLT1+443DDXNbcNRsltlgCQ6+oXgsrE2Pl5OhV73ZI/oM5Nj
+37cWEaHpXPiHdr1W0cD3aJ5qJ55TfTRkHP9xcMNQJHCn6aWPEHpR7xvvXW9WbJWfShnE2Xdvmw==
+"""
+
+COMMAND_OFF = """FlO5i3DRcrUmu2ZwIIv8b68EisGu8VCuqfGOydaR+xCA0n3f2W/EcqVj8MurRBFXYTrZ/uwa1W26
+ftCfvhdXNebBRwHr9Rj3id4bVfltJ8eT5/R3xY8kputklW2mrw9UfdISzAJqOPp9KZcU4K9p8g==
+"""
+
+class TapoP115Controller:
+    def __init__(self, device_ip):
+        self.device_ip = device_ip
+        self.session_id = None
+        self.token = None
+    
+    def perform_handshake(self):
+        """Führt den ersten Handshake durch und speichert die Session-ID"""
+        handshake_data = {
+            "method": "handshake",
+            "params": {
+                "key": PUBLIC_KEY
+            },
+            "requestTimeMils": 0
+        }
+        
+        headers = {
+            "Referer": f"http://{self.device_ip}:80",
+            "Accept": "application/json",
+            "requestByApp": "true",
+            "Content-Type": "application/json; charset=UTF-8"
+        }
+        
+        response = requests.post(
+            f"http://{self.device_ip}/app",
+            json=handshake_data,
+            headers=headers
+        )
+        
+        if response.status_code == 200:
+            data = response.json()
+            if data["error_code"] == 0:
+                # Session-ID aus dem Cookie extrahieren
+                self.session_id = response.cookies.get("TP_SESSIONID")
+                print(f"Handshake erfolgreich, Session-ID: {self.session_id}")
+                
+                # In einem echten Szenario würden wir hier den verschlüsselten Schlüssel entschlüsseln
+                # Da wir keinen privaten Schlüssel haben, speichern wir nur die Antwort
+                encrypted_key = data["result"]["key"]
+                print(f"Verschlüsselter Schlüssel: {encrypted_key}")
+                return True
+        
+        print("Handshake fehlgeschlagen")
+        return False
+    
+    def send_command(self, encrypted_command):
+        """Sendet einen vorbereiteten verschlüsselten Befehl"""
+        if not self.session_id:
+            print("Keine Session-ID. Bitte zuerst Handshake durchführen.")
+            return None
+        
+        # Token aus der Wireshark-Aufnahme (könnte sich ändern, oder vom Gerät abhängen)
+        token = "9DFAC92C53CEC92E67A9CB2E00B3CB2F"
+        
+        secure_data = {
+            "method": "securePassthrough",
+            "params": {
+                "request": encrypted_command
+            }
+        }
+        
+        headers = {
+            "Referer": f"http://{self.device_ip}:80",
+            "Accept": "application/json",
+            "requestByApp": "true",
+            "Content-Type": "application/json; charset=UTF-8",
+            "Cookie": f"TP_SESSIONID={self.session_id}"
+        }
+        
+        response = requests.post(
+            f"http://{self.device_ip}/app?token={token}",
+            json=secure_data,
+            headers=headers
+        )
+        
+        if response.status_code == 200:
+            data = response.json()
+            if data["error_code"] == 0:
+                # In einem echten Szenario würden wir die Antwort entschlüsseln
+                encrypted_response = data["result"]["response"]
+                print("Befehl erfolgreich gesendet")
+                return encrypted_response
+        
+        print("Fehler beim Senden des Befehls")
+        return None
+    
+    def turn_on(self):
+        """Schaltet die Steckdose ein"""
+        return self.send_command(COMMAND_ON)
+    
+    def turn_off(self):
+        """Schaltet die Steckdose aus"""
+        return self.send_command(COMMAND_OFF)
+
+# Verwendungsbeispiel
+if __name__ == "__main__":
+    controller = TapoP115Controller("192.168.0.102")
+    
+    # Handshake durchführen
+    if controller.perform_handshake():
+        # Steckdose einschalten
+        controller.turn_on()
+        
+        # Kurze Pause (im echten Code mit time.sleep)
+        print("Steckdose ist eingeschaltet")
+        
+        # Steckdose ausschalten
+        controller.turn_off()
+        print("Steckdose ist ausgeschaltet")

backend/tapo.py

@@ -0,0 +1,9 @@
+from PyP100 import PyP100
+
+p100 = PyP100.P100("192.168.0.102", "till.tomczak@mercedes-benz.com", "Agent045") #Creates a P100 plug object
+
+p100.handshake() #Creates the cookies required for further methods
+p100.login() #Sends credentials to the plug and creates AES Key and IV for further methods
+
+p100.turnOn() #Turns the connected plug on
+p100.turnOff() #Turns the connected plug off