test14.py

import socket
import hashlib
import re

PROJECTOR_IP = "10.168.222.248"
PASSWORD = "00000"
USERNAME = "EPSONWEB"

def md5(data):
    return hashlib.md5(data.encode()).hexdigest()

def http_request_with_digest(method, path, username=USERNAME, password=PASSWORD):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(5)
    sock.connect((PROJECTOR_IP, 80))
    request = f"{method} {path} HTTP/1.1\r\nHost: {PROJECTOR_IP}\r\nConnection: keep-alive\r\n\r\n"
    sock.send(request.encode())

    response = b""
    while True:
        try:
            chunk = sock.recv(4096)
            if not chunk:
                break
            response += chunk
            if b"\r\n\r\n" in response and b"401" in response[:100]:
                break
        except:
            break

    response_str = response.decode('utf-8', errors='ignore')
    auth_match = re.search(r'WWW-Authenticate: Digest (.+)', response_str)
    if not auth_match:
        sock.close()
        return response

    auth_params = {}
    for match in re.finditer(r'(\w+)="?([^",]+)"?', auth_match.group(1)):
        auth_params[match.group(1)] = match.group(2)

    realm = auth_params.get('realm', '')
    nonce = auth_params.get('nonce', '')
    qop = auth_params.get('qop', '')
    ha1 = md5(f"{username}:{realm}:{password}")
    ha2 = md5(f"{method}:{path}")
    nc = "00000001"
    cnonce = "0a4f113b"
    response_hash = md5(f"{ha1}:{nonce}:{nc}:{cnonce}:{qop}:{ha2}") if qop else md5(f"{ha1}:{nonce}:{ha2}")
    auth_header = f'Digest username="{username}", realm="{realm}", nonce="{nonce}", uri="{path}", '
    if qop:
        auth_header += f'qop={qop}, nc={nc}, cnonce="{cnonce}", '
    auth_header += f'response="{response_hash}"'

    sock.close()
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(5)
    sock.connect((PROJECTOR_IP, 80))
    request = f"{method} {path} HTTP/1.1\r\nHost: {PROJECTOR_IP}\r\nAuthorization: {auth_header}\r\nConnection: close\r\n\r\n"
    sock.send(request.encode())

    response = b""
    while True:
        try:
            chunk = sock.recv(4096)
            if not chunk:
                break
            response += chunk
        except:
            break
    sock.close()
    return response

# Test: Activer la source LAN
print("=== Test 1: Activation de la source LAN (KEY=53) ===")
resp = http_request_with_digest("GET", "/cgi-bin/directsend?KEY=53")
status = resp.split(b"\r\n")[0].decode() if resp else "No response"
print(f"Status: {status}")
if b"\r\n\r\n" in resp:
    content = resp.split(b"\r\n\r\n", 1)[1][:500]
    print(f"Content: {content.decode('utf-8', errors='ignore')}")

# Test: Query json pour voir l'état
print("\n\n=== Test 2: Interroger l'état avec json_query ===")
queries = [
    "SOURCE?",
    "DIRECTSEND?",
    "INFO?",
    "POWER?",
]

for query in queries:
    print(f"\nQuery: {query}")
    resp = http_request_with_digest("GET", f"/cgi-bin/json_query?jsoncallback={query}")
    if b"200 OK" in resp and b"\r\n\r\n" in resp:
        content = resp.split(b"\r\n\r\n", 1)[1][:500]
        print(f"  Response: {content.decode('utf-8', errors='ignore')}")

# Test: Chercher sur le port 3620/3621 (Epson iProjection)
print("\n\n=== Test 3: Vérification des ports Epson ===")
ports = [3620, 3621, 3629]
for port in ports:
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(2)
        result = sock.connect_ex((PROJECTOR_IP, port))
        if result == 0:
            print(f"Port {port}: OUVERT")
            # Essayer d'envoyer quelque chose
            sock.send(b"TEST\n")
            try:
                data = sock.recv(1024)
                print(f"  Réponse: {data[:100]}")
            except:
                print(f"  Pas de réponse")
        else:
            print(f"Port {port}: fermé")
        sock.close()
    except Exception as e:
        print(f"Port {port}: erreur - {e}")