#!/usr/bin/env bash set -euo pipefail die() { echo "fwd1: $*" >&2 exit 1 } command -v python3 >/dev/null 2>&1 || die "python3 is required; no compiler or third-party Python packages are used" python3 - "$@" <<'PY' import ipaddress import os import stat import struct import sys import tempfile from pathlib import Path ELF_HEADER = struct.Struct("<16sHHIQQQIHHHHHH") PROGRAM_HEADER = struct.Struct(" len(configured): raise PatchError("embedded hotfix configuration is truncated") magic, version, flags, size, ipv4, port, reserved = ( HOTFIX_CONFIG.unpack_from(configured, offset) ) if magic != HOTFIX_CONFIG_MAGIC or version != 1 or size != HOTFIX_CONFIG.size: raise PatchError("embedded hotfix configuration is incompatible") flags &= ~(HOTFIX_CONFIG_MANUAL_IP | HOTFIX_CONFIG_MANUAL_PORT) ipv4 = b"\0" * 4 port = 0 if manual_ip is not None: flags |= HOTFIX_CONFIG_MANUAL_IP ipv4 = manual_ip.packed if manual_port is not None: flags |= HOTFIX_CONFIG_MANUAL_PORT port = manual_port HOTFIX_CONFIG.pack_into( configured, offset, magic, version, flags, size, ipv4, port, reserved, ) return bytes(configured) def align_up(value, alignment): return (value + alignment - 1) & -alignment def signed_rel32(source_after_instruction, target): displacement = target - source_after_instruction if not -(1 << 31) <= displacement < (1 << 31): raise PatchError("original DT_INIT is outside x86-64 rel32 range") return struct.pack(" len(data): raise PatchError("program-header table extends past end of file") headers = [ list(PROGRAM_HEADER.unpack_from(data, phoff + i * phentsize)) for i in range(phnum) ] if not any(ph[0] == PT_LOAD for ph in headers): raise PatchError("ELF has no PT_LOAD segment") return header, headers def uses_ibt(data, headers): for note in (ph for ph in headers if ph[0] == PT_NOTE): offset, size = note[2], note[5] end = offset + size if end > len(data): raise PatchError("invalid PT_NOTE file range") position = offset while position + 12 <= end: name_size, description_size, note_type = struct.unpack_from( " end or description_end > end: raise PatchError("invalid PT_NOTE record") position = align_up(description_end, 4) if ( note_type != NT_GNU_PROPERTY_TYPE_0 or data[name_start:name_end] != b"GNU\0" ): continue property_offset = description_start while property_offset + 8 <= description_end: property_type, property_size = struct.unpack_from( " description_end: raise PatchError("invalid GNU property note") if ( property_type == GNU_PROPERTY_X86_FEATURE_1_AND and property_size >= 4 and struct.unpack_from(" len(data) or size < DYNAMIC_ENTRY.size or size % DYNAMIC_ENTRY.size ): raise PatchError("invalid PT_DYNAMIC file range") for entry_offset in range(offset, offset + size, DYNAMIC_ENTRY.size): tag, value = DYNAMIC_ENTRY.unpack_from(data, entry_offset) if tag == DT_INIT: return entry_offset, value or None if tag == DT_NULL: next_offset = entry_offset + DYNAMIC_ENTRY.size if next_offset + DYNAMIC_ENTRY.size <= offset + size: next_tag, _ = DYNAMIC_ENTRY.unpack_from(data, next_offset) if next_tag == DT_NULL: return entry_offset, None break raise PatchError("ELF has no DT_INIT and no spare dynamic-table entry") def append_patch_segment(data, header, headers, blob_size): page_size = 0x1000 segment_offset = align_up(len(data), page_size) highest_vaddr = max( ph[3] + ph[6] for ph in headers if ph[0] == PT_LOAD ) segment_vaddr = align_up(highest_vaddr, page_size) relocated_headers = [ph[:] for ph in headers] relocated_phnum = len(relocated_headers) + 1 relocated_phsize = relocated_phnum * PROGRAM_HEADER.size patch_offset = align_up(segment_offset + relocated_phsize, 16) patch_vaddr = segment_vaddr + patch_offset - segment_offset segment_size = patch_offset + blob_size - segment_offset new_header = list(header) new_header[5] = segment_offset new_header[10] = relocated_phnum new_segment = [ PT_LOAD, PF_R | PF_X, segment_offset, segment_vaddr, segment_vaddr, segment_size, segment_size, page_size, ] for ph in relocated_headers: if ph[0] == PT_PHDR: ph[2] = segment_offset ph[3] = segment_vaddr ph[4] = segment_vaddr ph[5] = relocated_phsize ph[6] = relocated_phsize return ( new_header, relocated_headers + [new_segment], patch_offset, patch_vaddr, ) def patch_bytes(data): if MARKER in data: raise PatchError("file is already patched by fwd1") header, headers = parse_headers(data) if uses_ibt(data, headers): raise PatchError( "ELF enables x86 IBT, which the hotfix wrapper does not support" ) init_slot, original_init = dynamic_init_slot(data, headers) provisional_stub = make_stub(0, None) extra_jump = 4 if original_init is not None else 0 blob_size = len(MARKER) + len(provisional_stub) + extra_jump new_header, relocated_headers, patch_offset, marker_vaddr = ( append_patch_segment(data, header, headers, blob_size) ) stub_vaddr = marker_vaddr + len(MARKER) patch_blob = MARKER + make_stub(stub_vaddr, original_init) relocated_phoff = new_header[5] if len(data) < relocated_phoff: data.extend(b"\0" * (relocated_phoff - len(data))) data.extend(b"\0" * (patch_offset + len(patch_blob) - len(data))) ELF_HEADER.pack_into(data, 0, *new_header) for index, ph in enumerate(relocated_headers): PROGRAM_HEADER.pack_into( data, relocated_phoff + index * PROGRAM_HEADER.size, *ph ) data[patch_offset : patch_offset + len(patch_blob)] = patch_blob DYNAMIC_ENTRY.pack_into(data, init_slot, DT_INIT, stub_vaddr) return data def patch_file(source, output): in_place = source.absolute() == output.absolute() if in_place and source.is_symlink(): raise PatchError( "in-place patching through a symlink is not supported; " "use the symlink target" ) source_stat = source.stat() if not stat.S_ISREG(source_stat.st_mode): raise PatchError("input is not a regular file") if in_place and source_stat.st_nlink > 1: raise PatchError( f"input has {source_stat.st_nlink} hard links; " "use a separate output path" ) patched = patch_bytes(bytearray(source.read_bytes())) output_dir = output.parent.resolve() output_dir.mkdir(parents=True, exist_ok=True) fd, temporary_name = tempfile.mkstemp(prefix=f".{output.name}.", dir=output_dir) temporary = Path(temporary_name) try: with os.fdopen(fd, "wb") as stream: stream.write(patched) stream.flush() os.fsync(stream.fileno()) os.chmod(temporary, stat.S_IMODE(source_stat.st_mode)) os.replace(temporary, output) finally: try: temporary.unlink() except FileNotFoundError: pass try: source, output, manual_ip, manual_port = parse_cli(sys.argv[1:]) HOTFIX = configure_hotfix(HOTFIX, manual_ip, manual_port) patch_file(source, output) except (OSError, PatchError) as error: print(f"fwd1: {error}", file=sys.stderr) raise SystemExit(1) PY