"""Automatic ISO -> XEX/GOD pipeline; no console-side patching of executables.""" import asyncio import json import os import re import shutil import tarfile import zipfile from pathlib import Path, PurePosixPath from . import store from .common import is_archive, is_additional_volume, safe_title READY = store.DATA / 'ready' WORK = store.DATA / 'work' RAW = store.DATA / 'raw' LOGS = store.DATA / 'logs' class ProcessingError(RuntimeError): pass def safe_archive_path(name): # Archive members are always relative, independent of platform. norm = name.replace('\\', '/') if (not norm or norm.startswith('/') or '\x00' in norm or re.match(r'^[A-Za-z]:', norm) or any(x == '..' for x in PurePosixPath(norm).parts)): raise ProcessingError(f'Archiwum zawiera niebezpieczna sciezke: {name[:150]}') def verify_no_symlinks(directory): for root, dirs, files in os.walk(directory, followlinks=False): for name in dirs + files: path = Path(root) / name if path.is_symlink(): raise ProcessingError('Pliki-symboliczne linki nie sa dopuszczalne: ' + name) async def command(job_id, *args): LOGS.mkdir(parents=True, exist_ok=True) log_path = LOGS / f'{job_id}.log' with log_path.open('ab') as log: log.write(('\n>>> ' + ' '.join(str(x) for x in args) + '\n').encode('utf-8')) process = await asyncio.create_subprocess_exec(*map(str, args), stdout=log, stderr=log, stdin=asyncio.subprocess.DEVNULL) try: code = await process.wait() except asyncio.CancelledError: process.terminate() await process.wait() raise if code: data = log_path.read_bytes()[-2500:].decode('utf-8', errors='replace') raise ProcessingError(f'Narzedzie {args[0]} zakonczone bledem {code}. Ostatni log:\n{data[-1400:]}') def list_primary_archives(source): files = [p for p in source.rglob('*') if p.is_file() and not p.is_symlink()] archives = [p for p in files if is_archive(p.name) and not is_additional_volume(p.name)] # Multipart .rar is detected by the first volume, subsequent parts are ignored. # A single .zip with .z01 volumes is also supported by 7z. return archives async def unpack_archive(job_id, archive, destination): destination.mkdir(parents=True, exist_ok=True) lower = archive.name.lower() store.update(job_id, state='extracting', note='Rozpakowywanie ' + archive.name[:100]) if lower.endswith('.zip') and not any(archive.parent.glob(archive.stem + '.z[0-9][0-9]')): # Standard ZIP: validate entries to prevent writes outside destination. with zipfile.ZipFile(archive) as z: for member in z.infolist(): safe_archive_path(member.filename) if (member.external_attr >> 16) & 0o170000 == 0o120000: raise ProcessingError('Archiwum ZIP zawiera link symboliczny') await asyncio.to_thread(z.extractall, destination) elif lower.endswith(('.tar', '.tar.gz', '.tgz', '.tar.xz', '.tar.bz2')): # tarfile's filters are version-dependent; reject links/special entries ourselves. with tarfile.open(archive) as tf: for member in tf.getmembers(): safe_archive_path(member.name) if not (member.isfile() or member.isdir()): raise ProcessingError('Archiwum TAR zawiera link lub plik specjalny') await asyncio.to_thread(tf.extractall, destination, filter='data') else: # 7z/RAR/multipart. Validate file paths *before* invoking extraction. listing = await asyncio.create_subprocess_exec('7z','l','-slt',str(archive), stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE) listing_out, listing_err = await listing.communicate() if listing.returncode: if not lower.endswith('.rar'): raise ProcessingError('7z nie rozpoznaje archiwum: ' + listing_err[-300:].decode(errors='replace')) # RAR5 fallback: libunarr (unar) understands many archives missing # from Debian's p7zip package. Validate lsar entries first. lsar = await asyncio.create_subprocess_exec('lsar', '-json', '-no-recursion', str(archive), stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE) stdout, stderr = await lsar.communicate() if lsar.returncode: raise ProcessingError('7z/unar nie rozpoznaje RAR: ' + stderr[-400:].decode(errors='replace')) try: items = json.loads(stdout).get('lsarContents', []) if not items: raise ProcessingError('Nie mozna odczytac zawartosci RAR') for item in items: safe_archive_path(item['XADFileName']) if any(item.get(x) for x in ('XADIsLink', 'XADIsSymbolicLink', 'XADLinkDestination')): raise ProcessingError('RAR zawiera link symboliczny') except (ValueError, KeyError, TypeError) as exc: raise ProcessingError('Niepoprawna lista plikow RAR') from exc await command(job_id, 'unar', '-no-directory', '-force-overwrite', '-output-directory', str(destination), str(archive)) verify_no_symlinks(destination) return data = listing_out.decode('utf-8', errors='replace') entries = data.split('----------', 1) if len(entries) != 2: raise ProcessingError('Nie mozna zweryfikowac listy plikow archiwum') for line in entries[1].splitlines(): if line.startswith('Path = '): safe_archive_path(line[7:]) if line.startswith(('Symbolic Link = ', 'Hard Link = ')): raise ProcessingError('Archiwum zawiera link symboliczny lub twardy') await command(job_id, '7z', 'x', '-y', '-bd', str(archive), f'-o{destination}') verify_no_symlinks(destination) async def locate_payload(job_id, root, tmp): """Find X360 ISOs or default.xex, recursively extracting up to 2 archive layers.""" current = root for depth in range(3): verify_no_symlinks(current) isos = sorted(p for p in current.rglob('*') if p.is_file() and p.suffix.lower() == '.iso') xex_files = sorted(p for p in current.rglob('*') if p.is_file() and p.name.lower() == 'default.xex') if isos or xex_files: return isos, xex_files archives = list_primary_archives(current) if not archives: break if len(archives) != 1: raise ProcessingError('Znaleziono kilka niezaleznych archiwow. Dodaj jedno na zadanie lub polacz czesci gry w jedno archiwum.') if depth >= 2: raise ProcessingError('Archiwa zagniezdzone glebiej niz 3 poziomy nie sa obsługiwane') out = tmp / f'unpacked-{depth}' await unpack_archive(job_id, archives[0], out) current = out raise ProcessingError('Brak obrazu .iso ani default.xex. Sprawdz czy pobrano wszystkie czesci archiwum.') async def process_job(job): jid = job['id'] fmt = job['format'] source = RAW / jid temp = WORK / jid slug = safe_title(job['title'])[:55] + '-' + jid[:8] dest = READY / fmt / slug atomic_dest = READY / fmt / ('.' + slug + '.working') if not source.exists(): raise ProcessingError('Brak pobranych plikow w katalogu raw') temp.mkdir(parents=True, exist_ok=True) if atomic_dest.exists(): shutil.rmtree(atomic_dest) atomic_dest.mkdir(parents=True, exist_ok=True) store.update(jid, state='preparing', note='Analiza pobranych plikow', speed=0) try: verify_no_symlinks(source) if fmt == 'raw': await asyncio.to_thread(shutil.copytree, source, atomic_dest / 'files') else: isos, xex_files = await locate_payload(jid, source, temp) if fmt == 'xex': store.update(jid, state='converting', note='Tworzenie folderu XEX') if xex_files: if len(xex_files) > 1: raise ProcessingError('Kilka plikow default.xex; wybierz jeden obraz ISO na zadanie') await asyncio.to_thread(shutil.copytree, xex_files[0].parent, atomic_dest / 'Gra') else: for n, iso in enumerate(isos, 1): dest_dir = atomic_dest / ('Gra' if len(isos) == 1 else f'Disc_{n}') await command(jid, 'xdvdfs', 'unpack', str(iso), str(dest_dir)) if not any(p.name.lower() == 'default.xex' for p in dest_dir.rglob('*') if p.is_file()): raise ProcessingError(f'Obraz ISO nie zawiera default.xex: {iso.name}') else: store.update(jid, state='converting', note='Konwersja ISO do GOD') if not isos and xex_files: if len(xex_files) > 1: raise ProcessingError('Kilka folderow default.xex; GOD wymaga jednego obrazu') packed = temp / 'packed.xiso.iso' await command(jid, 'xdvdfs', 'pack', str(xex_files[0].parent), str(packed)) isos = [packed] for n, iso in enumerate(isos, 1): # iso2god creates // inside provided root. # Prefix these folders with Content/0000000000000000 for Xbox internal HDD. disc_dir = atomic_dest if len(isos) == 1 else atomic_dest / f'Disc_{n}' content = disc_dir / 'Content' / '0000000000000000' content.mkdir(parents=True, exist_ok=True) await command(jid, 'iso2god', str(iso), str(content)) if not any(p.is_dir() and re.fullmatch(r'[0-9A-Fa-f]{8}', p.name) for p in content.iterdir()): raise ProcessingError(f'Brak katalogu TitleID w wyniku ISO2GOD: {iso.name}') if dest.exists(): raise ProcessingError('Folder wynikowy juz istnieje: ' + str(dest)) atomic_dest.rename(dest) store.update(jid, state='done', progress=100, output=str(dest), note='Gotowe do skopiowania na Xbox 360') finally: # Full download is preserved; only temporary extraction is deleted. shutil.rmtree(temp, ignore_errors=True) if atomic_dest.exists(): shutil.rmtree(atomic_dest, ignore_errors=True)