251 lines
10 KiB
Python
Executable File
251 lines
10 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Client de l'agent de l'iPod émulé (emuagent, via ipod-netd sur 127.0.0.1:7788).
|
|
|
|
ipod-agent ping
|
|
ipod-agent apps bundle-id, nom et chemin des apps installées
|
|
ipod-agent launch BUNDLE_ID [--log F] [--url URL]
|
|
lance une app (stdout/stderr, donc NSLog, dans F sur l'iPod ; URL à ouvrir)
|
|
ipod-agent fetch http://hôte/chemin requête HTTP depuis l'iPod (test réseau : DNS + TCP)
|
|
ipod-agent front app au premier plan
|
|
ipod-agent screen | wake état du verrouillage / réveil sans clavier de code (agent v11)
|
|
ipod-agent unlock réveille puis glisse et vérifie le déverrouillage (sans code)
|
|
ipod-agent exec PROG [ARGS...] exécute un programme (pas de shell : chemin absolu)
|
|
ipod-agent ps | net | respring | sync
|
|
ipod-agent kill PID|NOM [SIGNAL]
|
|
ipod-agent ls CHEMIN | rm CHEMIN | mkdir CHEMIN
|
|
ipod-agent get CHEMIN_IPOD [FICHIER_LOCAL]
|
|
ipod-agent put FICHIER_LOCAL CHEMIN_IPOD [MODE]
|
|
ipod-agent push-app MonApp.app copie l'app, synchronise et redémarre l'iPod (NAND conservée)
|
|
ipod-agent wait [SECONDES] attend que l'agent réponde (réseau PPP monté)
|
|
|
|
Code de sortie : 0 si OK, 1 si erreur, code de sortie du programme pour exec.
|
|
"""
|
|
import os, socket, subprocess, sys, tempfile, time
|
|
|
|
HOST = os.environ.get("IPOD_AGENT_HOST", "127.0.0.1")
|
|
PORT = int(os.environ.get("IPOD_AGENT_PORT", "7788"))
|
|
|
|
|
|
class Agent:
|
|
def __init__(self, timeout=330):
|
|
self.s = socket.create_connection((HOST, PORT), timeout=timeout)
|
|
self.f = self.s.makefile("rb")
|
|
|
|
def call(self, *fields, data=None, sink=None):
|
|
line = "\t".join(str(x) for x in fields) + "\n"
|
|
self.s.sendall(line.encode())
|
|
if data is not None:
|
|
if hasattr(data, "read"):
|
|
remaining = int(fields[-1])
|
|
while remaining:
|
|
chunk = data.read(min(65536, remaining))
|
|
if not chunk:
|
|
raise ConnectionError("fichier local tronqué pendant l'upload")
|
|
self.s.sendall(chunk)
|
|
remaining -= len(chunk)
|
|
else:
|
|
self.s.sendall(data)
|
|
hdr = self.f.readline(4096).decode().rstrip("\n")
|
|
if not hdr:
|
|
raise ConnectionError("connexion fermée par l'agent")
|
|
parts = hdr.split(" ", 2)
|
|
if len(parts) < 2:
|
|
raise ConnectionError("en-tête invalide")
|
|
n = int(parts[1])
|
|
if n < 0 or parts[0] not in ("OK", "ERR"):
|
|
raise ConnectionError("en-tête invalide")
|
|
body = bytearray()
|
|
received = 0
|
|
while received < n:
|
|
chunk = self.f.read(min(65536, n - received))
|
|
if not chunk:
|
|
raise ConnectionError(f"réponse tronquée : {received} octets reçus sur {n}")
|
|
received += len(chunk)
|
|
if sink is not None and parts[0] == "OK":
|
|
sink.write(chunk)
|
|
else:
|
|
body.extend(chunk)
|
|
body = bytes(body)
|
|
return parts[0] == "OK", body, parts[2] if len(parts) > 2 else None
|
|
|
|
|
|
def out(body):
|
|
sys.stdout.buffer.write(body)
|
|
if body and not body.endswith(b"\n"):
|
|
sys.stdout.buffer.write(b"\n")
|
|
|
|
|
|
def emu_home():
|
|
tools_home = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
default_home = tools_home if os.path.isdir(os.path.join(tools_home, "images")) else os.path.expanduser("~/ipod-emu")
|
|
return os.environ.get("IPOD_EMU_HOME", default_home)
|
|
|
|
|
|
def unlock(a):
|
|
def screen():
|
|
ok, body, _ = a.call("screen")
|
|
if not ok:
|
|
sys.exit("État d'écran indisponible (agent v11 requis) : " + body.decode())
|
|
return dict(field.split("=", 1) for field in body.decode().split())
|
|
|
|
state = screen()
|
|
if state.get("locked") == "0":
|
|
print("Écran déverrouillé")
|
|
return
|
|
if state.get("passcode") != "0":
|
|
sys.exit("Un code de verrouillage est activé : déverrouiller manuellement.")
|
|
# Le pilote tactile doit avoir terminé son démarrage avant tout glissement.
|
|
try:
|
|
offset_path = os.path.join(emu_home(), "run", "boot-log-offset")
|
|
offset = 0
|
|
if os.path.exists(offset_path):
|
|
with open(offset_path) as f:
|
|
offset = int(f.read())
|
|
with open(os.path.join(emu_home(), "run", "serial.log"), "rb") as f:
|
|
f.seek(offset)
|
|
serial = f.read()
|
|
except OSError as e:
|
|
sys.exit(f"Console de session inaccessible : {e}")
|
|
helper_ready = any(b"SCHelper" in line and b'"EXIT"' in line for line in serial.splitlines())
|
|
if serial.count(b"AppleMultitouchSPIUserClient::attach") < 2 or not helper_ready:
|
|
sys.exit("Contrôleur tactile pas encore prêt : attendre ipod-emu wait.")
|
|
qmp = os.path.join(os.path.dirname(os.path.abspath(__file__)), "ipod-qmp")
|
|
sock = os.environ.get("IPOD_QMP", os.path.join(emu_home(), "run", "qmp.sock"))
|
|
for attempt in range(3):
|
|
ok, body, _ = a.call("wake")
|
|
if not ok:
|
|
sys.exit(body.decode())
|
|
time.sleep(2) # laisser le pilote réactiver l'écran et le tactile
|
|
# Terminer le geste au bord droit du curseur de déverrouillage.
|
|
subprocess.run([qmp, "-s", sock, "swipe", "60", "450", "319", "450"], check=True)
|
|
time.sleep(2)
|
|
state = screen()
|
|
if state.get("locked") == "0":
|
|
print("Écran déverrouillé")
|
|
return
|
|
if state.get("passcode") != "0":
|
|
sys.exit("Un code de verrouillage est activé : déverrouiller manuellement.")
|
|
sys.exit("Écran toujours verrouillé après trois glissements.")
|
|
|
|
|
|
def push_app(a, app):
|
|
home = emu_home()
|
|
run = os.path.join(home, "run")
|
|
try:
|
|
with open(os.path.join(run, "nand", ".ipod-nand-format")) as f:
|
|
layout = f.read().strip()
|
|
with open(os.path.join(run, "fw")) as f:
|
|
firmware = f.read().strip()
|
|
with open(os.path.join(home, "images", firmware, "machine-opts")) as f:
|
|
machine_opts = f.read()
|
|
except OSError:
|
|
sys.exit("push-app nécessite une image locale avec persistance NAND (ipod-mkimage / ipod-build-fw).")
|
|
if layout != "vsvfl-production-v1" or ",nand-write=on" not in machine_opts:
|
|
sys.exit("Écritures NAND désactivées : reconstruire l'image avant push-app.")
|
|
app = app.rstrip("/")
|
|
dest = "/Applications/" + os.path.basename(app)
|
|
a.call("rm", dest)
|
|
n = 0
|
|
for root, dirs, files in os.walk(app):
|
|
for f in files:
|
|
p = os.path.join(root, f)
|
|
rel = os.path.relpath(p, app)
|
|
st = os.stat(p)
|
|
with open(p, "rb") as fh:
|
|
data = fh.read()
|
|
mode = "755" if (st.st_mode & 0o111 or data[:4] in (b"\xce\xfa\xed\xfe", b"\xca\xfe\xba\xbe")) else "644"
|
|
ok, body, _ = a.call("put", f"{dest}/{rel}", mode, len(data), data=data)
|
|
if not ok:
|
|
sys.exit(body.decode())
|
|
n += 1
|
|
# SpringBoard ne relit la liste des apps qu'à son démarrage
|
|
a.call("rm", "/private/var/mobile/Library/Caches/com.apple.mobile.installation.plist")
|
|
ok, body, _ = a.call("sync")
|
|
if not ok:
|
|
sys.exit(body.decode())
|
|
# Le cache d'installation iOS 2.x peut bloquer SpringBoard lors d'une
|
|
# relance seule. Un démarrage complet le reconstruit, sans refaire l'image.
|
|
a.f.close()
|
|
a.s.close()
|
|
launcher = os.path.join(os.path.dirname(os.path.abspath(__file__)), "ipod-emu")
|
|
print(f"{n} fichiers copiés dans {dest}, redémarrage de l'iPod", flush=True)
|
|
subprocess.run([launcher, "stop"], check=True)
|
|
subprocess.run([launcher, "start", "-w", firmware, "-W"], check=True)
|
|
|
|
|
|
def main():
|
|
args = sys.argv[1:]
|
|
if not args or args[0] in ("-h", "--help"):
|
|
sys.exit(__doc__)
|
|
cmd, rest = args[0], args[1:]
|
|
|
|
if cmd == "wait":
|
|
limit = float(rest[0]) if rest else 300
|
|
t0 = time.time()
|
|
while True:
|
|
try:
|
|
ok, body, _ = Agent(timeout=5).call("ping")
|
|
print(body.decode(), f"({time.time() - t0:.0f} s)")
|
|
return
|
|
except OSError:
|
|
if time.time() - t0 > limit:
|
|
sys.exit("agent injoignable (ipod-netd lancé ? réseau PPP monté ?)")
|
|
time.sleep(2)
|
|
|
|
try:
|
|
a = Agent()
|
|
except OSError as e:
|
|
sys.exit(f"agent injoignable sur {HOST}:{PORT} ({e}) — ipod-netd lancé ?")
|
|
|
|
if cmd == "unlock":
|
|
a.s.settimeout(15)
|
|
unlock(a)
|
|
return
|
|
if cmd == "get":
|
|
tmp = None
|
|
try:
|
|
if len(rest) > 1:
|
|
destination = os.path.abspath(rest[1])
|
|
fd, tmp = tempfile.mkstemp(prefix=".ipod-get-", dir=os.path.dirname(destination))
|
|
with os.fdopen(fd, "wb") as fh:
|
|
ok, body, _ = a.call("get", rest[0], sink=fh)
|
|
fh.flush()
|
|
os.fsync(fh.fileno())
|
|
if ok:
|
|
os.replace(tmp, destination)
|
|
tmp = None
|
|
else:
|
|
ok, body, _ = a.call("get", rest[0], sink=sys.stdout.buffer)
|
|
if not ok:
|
|
sys.exit(body.decode())
|
|
finally:
|
|
if tmp is not None:
|
|
os.unlink(tmp)
|
|
return
|
|
if cmd == "put":
|
|
mode = rest[2] if len(rest) > 2 else ("755" if os.access(rest[0], os.X_OK) else "644")
|
|
with open(rest[0], "rb") as fh:
|
|
ok, body, _ = a.call("put", rest[1], mode, os.fstat(fh.fileno()).st_size, data=fh)
|
|
elif cmd == "push-app":
|
|
push_app(a, rest[0])
|
|
return
|
|
elif cmd == "launch":
|
|
fields = ["launch", rest[0], rest[rest.index("--log") + 1] if "--log" in rest else "-"]
|
|
if "--url" in rest:
|
|
fields.append(rest[rest.index("--url") + 1])
|
|
ok, body, _ = a.call(*fields)
|
|
else:
|
|
ok, body, info = a.call(cmd, *rest)
|
|
if cmd == "exec":
|
|
out(body)
|
|
sys.exit(int(info) if ok and info else (0 if ok else 1))
|
|
out(body)
|
|
sys.exit(0 if ok else 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except (OSError, ValueError, ConnectionError) as e:
|
|
sys.exit(f"Erreur de transfert ou de protocole : {e}")
|