debug du upnp mediaserver
This commit is contained in:
175
tools/discover_upnp.py
Executable file
175
tools/discover_upnp.py
Executable file
@@ -0,0 +1,175 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
UPnP Device Discovery Tool
|
||||
Envoie une requête M-SEARCH SSDP et collecte les réponses des devices
|
||||
"""
|
||||
|
||||
import socket
|
||||
import struct
|
||||
import time
|
||||
import sys
|
||||
from urllib.parse import urlparse
|
||||
from urllib.request import urlopen
|
||||
|
||||
SSDP_ADDR = "239.255.255.250"
|
||||
SSDP_PORT = 1900
|
||||
SSDP_MX = 3
|
||||
SSDP_ST = "ssdp:all"
|
||||
|
||||
M_SEARCH = f"""M-SEARCH * HTTP/1.1
|
||||
HOST: {SSDP_ADDR}:{SSDP_PORT}
|
||||
MAN: "ssdp:discover"
|
||||
MX: {SSDP_MX}
|
||||
ST: {SSDP_ST}
|
||||
USER-AGENT: PMOMusic UPnP Discovery Tool
|
||||
|
||||
"""
|
||||
|
||||
def discover_upnp_devices(timeout=5):
|
||||
"""Découvre les devices UPnP sur le réseau local"""
|
||||
|
||||
devices = {}
|
||||
|
||||
# Créer le socket UDP
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
|
||||
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
sock.settimeout(timeout)
|
||||
|
||||
# Envoyer la requête M-SEARCH
|
||||
print(f"🔍 Envoi de la requête M-SEARCH sur {SSDP_ADDR}:{SSDP_PORT}...")
|
||||
print(f"⏱️ Timeout: {timeout}s\n")
|
||||
|
||||
message = M_SEARCH.replace('\n', '\r\n').encode('utf-8')
|
||||
sock.sendto(message, (SSDP_ADDR, SSDP_PORT))
|
||||
|
||||
# Collecter les réponses
|
||||
start_time = time.time()
|
||||
|
||||
while time.time() - start_time < timeout:
|
||||
try:
|
||||
data, addr = sock.recvfrom(65507)
|
||||
response = data.decode('utf-8', errors='ignore')
|
||||
|
||||
# Parser la réponse
|
||||
location = None
|
||||
server = None
|
||||
st = None
|
||||
usn = None
|
||||
|
||||
for line in response.split('\r\n'):
|
||||
if line.lower().startswith('location:'):
|
||||
location = line.split(':', 1)[1].strip()
|
||||
elif line.lower().startswith('server:'):
|
||||
server = line.split(':', 1)[1].strip()
|
||||
elif line.lower().startswith('st:'):
|
||||
st = line.split(':', 1)[1].strip()
|
||||
elif line.lower().startswith('usn:'):
|
||||
usn = line.split(':', 1)[1].strip()
|
||||
|
||||
if location and location not in devices:
|
||||
devices[location] = {
|
||||
'location': location,
|
||||
'server': server,
|
||||
'st': st,
|
||||
'usn': usn,
|
||||
'from': addr[0]
|
||||
}
|
||||
|
||||
except socket.timeout:
|
||||
break
|
||||
except Exception as e:
|
||||
print(f"⚠️ Erreur lors de la réception: {e}")
|
||||
|
||||
sock.close()
|
||||
return devices
|
||||
|
||||
def fetch_device_description(location):
|
||||
"""Récupère la description XML du device"""
|
||||
try:
|
||||
response = urlopen(location, timeout=3)
|
||||
return response.read().decode('utf-8')
|
||||
except Exception as e:
|
||||
return f"Error: {e}"
|
||||
|
||||
def main():
|
||||
print("=" * 70)
|
||||
print(" 🔍 UPnP Device Discovery Tool")
|
||||
print("=" * 70)
|
||||
print()
|
||||
|
||||
devices = discover_upnp_devices(timeout=5)
|
||||
|
||||
# Filtrer pour ne garder que les MediaServers
|
||||
media_servers = {}
|
||||
for loc, info in devices.items():
|
||||
if 'MediaServer' in str(info.get('st', '')):
|
||||
media_servers[loc] = info
|
||||
|
||||
print(f"\n📊 Résultats:")
|
||||
print(f" Total devices trouvés: {len(devices)}")
|
||||
print(f" MediaServers trouvés: {len(media_servers)}\n")
|
||||
|
||||
if not media_servers:
|
||||
print("❌ Aucun MediaServer trouvé!\n")
|
||||
print("📋 Tous les devices trouvés:")
|
||||
for loc, info in devices.items():
|
||||
print(f"\n - Location: {loc}")
|
||||
print(f" ST: {info.get('st', 'N/A')}")
|
||||
print(f" Server: {info.get('server', 'N/A')}")
|
||||
return
|
||||
|
||||
# Analyser chaque MediaServer
|
||||
for idx, (location, info) in enumerate(media_servers.items(), 1):
|
||||
print("=" * 70)
|
||||
print(f"📡 MediaServer #{idx}")
|
||||
print("=" * 70)
|
||||
print(f"Location: {location}")
|
||||
print(f"From IP: {info['from']}")
|
||||
print(f"Server: {info.get('server', 'N/A')}")
|
||||
print(f"USN: {info.get('usn', 'N/A')}")
|
||||
print()
|
||||
|
||||
# Récupérer la description
|
||||
print("📄 Fetching device description...")
|
||||
desc = fetch_device_description(location)
|
||||
|
||||
# Analyser la description
|
||||
if desc and not desc.startswith("Error"):
|
||||
print("\n📝 Device Description XML:")
|
||||
print("-" * 70)
|
||||
# Afficher les premières lignes
|
||||
lines = desc.split('\n')
|
||||
for line in lines[:50]: # Limiter à 50 lignes
|
||||
print(line)
|
||||
if len(lines) > 50:
|
||||
print(f"... ({len(lines) - 50} more lines)")
|
||||
print("-" * 70)
|
||||
|
||||
# Extraire les infos importantes
|
||||
import re
|
||||
friendly_name = re.search(r'<friendlyName>([^<]+)</friendlyName>', desc)
|
||||
manufacturer = re.search(r'<manufacturer>([^<]+)</manufacturer>', desc)
|
||||
model_name = re.search(r'<modelName>([^<]+)</modelName>', desc)
|
||||
udn = re.search(r'<UDN>([^<]+)</UDN>', desc)
|
||||
|
||||
print("\n📋 Device Info:")
|
||||
if friendly_name:
|
||||
print(f" Friendly Name: {friendly_name.group(1)}")
|
||||
if manufacturer:
|
||||
print(f" Manufacturer: {manufacturer.group(1)}")
|
||||
if model_name:
|
||||
print(f" Model Name: {model_name.group(1)}")
|
||||
if udn:
|
||||
print(f" UDN: {udn.group(1)}")
|
||||
|
||||
# Vérifier le format de l'UDN
|
||||
udn_value = udn.group(1)
|
||||
if not udn_value.startswith('uuid:'):
|
||||
print(f" ⚠️ WARNING: UDN ne commence pas par 'uuid:' !")
|
||||
else:
|
||||
print(f"❌ Erreur lors de la récupération: {desc}")
|
||||
|
||||
print("\n")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user