LIEF/tests/elf/add_segment.py
2020-11-09 21:02:50 +01:00

218 lines
7.6 KiB
Python

#!/usr/bin/env python
import logging
import os
import re
import shutil
import stat
import subprocess
import sys
import tempfile
import unittest
from subprocess import Popen
from unittest import TestCase
import lief
from lief.ELF import Segment
from utils import get_sample, has_recent_glibc
lief.logging.set_level(lief.logging.LOGGING_LEVEL.INFO)
CURRENT_DIRECTORY = os.path.dirname(os.path.abspath(__file__))
class TestAddSegment(TestCase):
def setUp(self):
self.logger = logging.getLogger(__name__)
self.tmp_dir = tempfile.mkdtemp(suffix='_lief_test_add_segment')
self.logger.debug("temp dir: {}".format(self.tmp_dir))
@unittest.skipUnless(sys.platform.startswith("linux"), "requires Linux")
@unittest.skipUnless(has_recent_glibc(), "Need a recent GLIBC version")
def test_simple(self):
sample_path = get_sample('ELF/ELF64_x86-64_binary_ls.bin')
stub = lief.parse(os.path.join(CURRENT_DIRECTORY, "hello_lief.bin"))
output = os.path.join(self.tmp_dir, "ls.segment")
target = lief.parse(sample_path)
for i in range(4):
segment = stub.segments[0]
original_va = segment.virtual_address
segment.virtual_address = 0
segment = target.add(segment)
new_ep = (stub.header.entrypoint - original_va) + segment.virtual_address
target.header.entrypoint = new_ep
target.write(output)
st = os.stat(output)
os.chmod(output, st.st_mode | stat.S_IEXEC)
p = Popen(output, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
stdout, _ = p.communicate()
self.logger.debug(stdout.decode("utf8"))
self.assertIsNotNone(re.search(r'LIEF is Working', stdout.decode("utf8")))
@unittest.skipUnless(sys.platform.startswith("linux"), "requires Linux")
@unittest.skipUnless(has_recent_glibc(), "Need a recent GLIBC version")
def test_gcc(self):
sample_path = get_sample('ELF/ELF64_x86-64_binary_gcc.bin')
stub = lief.parse(os.path.join(CURRENT_DIRECTORY, "hello_lief.bin"))
output = os.path.join(self.tmp_dir, "gcc.segment")
target = lief.parse(sample_path)
segment = stub.segments[0]
original_va = segment.virtual_address
segment.virtual_address = 0
segment = target.add(segment)
new_ep = (stub.header.entrypoint - original_va) + segment.virtual_address
target.header.entrypoint = new_ep
target.write(output)
st = os.stat(output)
os.chmod(output, st.st_mode | stat.S_IEXEC)
p = Popen(output, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
stdout, _ = p.communicate()
self.logger.debug(stdout.decode("utf8"))
self.assertIsNotNone(re.search(r'LIEF is Working', stdout.decode("utf8")))
@unittest.skipUnless(sys.platform.startswith("linux"), "requires Linux")
def test_static(self):
sample_path = get_sample('ELF/ELF64_x86-64_binary_static-binary.bin')
stub = lief.parse(os.path.join(CURRENT_DIRECTORY, "hello_lief.bin"))
output = os.path.join(self.tmp_dir, "static.segment")
target = lief.parse(sample_path)
segment = stub.segments[0]
original_va = segment.virtual_address
segment.virtual_address = 0
segment = target.add(segment)
new_ep = (stub.header.entrypoint - original_va) + segment.virtual_address
target.header.entrypoint = new_ep
target.write(output)
st = os.stat(output)
os.chmod(output, st.st_mode | stat.S_IEXEC)
p = Popen(output, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
stdout, _ = p.communicate()
self.logger.debug(stdout.decode("utf8"))
self.assertIsNotNone(re.search(r'LIEF is Working', stdout.decode("utf8")))
@unittest.skipUnless(sys.platform.startswith("linux"), "requires Linux")
def test_misc(self):
list_binaries = [
'/usr/bin/ls',
'/usr/bin/ssh',
'/usr/bin/nm',
'/usr/bin/openssl',
'/usr/bin/bc',
'/usr/bin/bzip2',
'/usr/bin/cp',
'/usr/bin/find',
'/usr/bin/file',
]
for binary in list_binaries:
self.logger.debug("Test with '{}'".format(binary))
self.run_add_segment(binary)
def run_add_segment(self, target):
if not os.path.isfile(target):
return
name = os.path.basename(target)
stub = lief.parse(os.path.join(CURRENT_DIRECTORY, "hello_lief.bin"))
target = lief.parse(target)
output = os.path.join(self.tmp_dir, "{}.segment".format(name))
for i in range(6):
segment = stub.segments[0]
original_va = segment.virtual_address
segment.virtual_address = 0
segment = target.add(segment)
new_ep = (stub.header.entrypoint - original_va) + segment.virtual_address
target.header.entrypoint = new_ep
target.write(output)
st = os.stat(output)
os.chmod(output, st.st_mode | stat.S_IEXEC)
p = Popen(output, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
stdout, _ = p.communicate()
self.logger.debug(stdout.decode("utf8"))
self.assertIsNotNone(re.search(r'LIEF is Working', stdout.decode("utf8")))
@unittest.skipUnless(False, "requires Linux")
def test_libc(self):
stub = lief.parse(os.path.join(CURRENT_DIRECTORY, "hello_lief.bin"))
tmp_dir = tempfile.mkdtemp(suffix='_lief_test_add_segment_libc')
self.logger.debug("temp dir: {}".format(tmp_dir))
libc_name = "libc.so.6"
for e in lief.parse("/bin/ls").libraries:
if e.startswith("libc."):
libc_name = e
break;
self.logger.debug("libc used: {}".format(libc_name))
libc = lief.parse('/usr/lib/{}'.format(libc_name))
out = os.path.join(tmp_dir, libc_name)
for i in range(10):
segment = stub.segments[0]
original_va = segment.virtual_address
segment.virtual_address = 0
segment = libc.add(segment)
new_ep = (stub.header.entrypoint - original_va) + segment.virtual_address
if libc.has(lief.ELF.DYNAMIC_TAGS.INIT_ARRAY):
init_array = libc.get(lief.ELF.DYNAMIC_TAGS.INIT_ARRAY)
callbacks = init_array.array
callbacks[0] = new_ep
init_array.array = callbacks
if libc.has(lief.ELF.DYNAMIC_TAGS.INIT):
init = libc.get(lief.ELF.DYNAMIC_TAGS.INIT)
init.value = new_ep
libc.write(out)
st = os.stat(out)
os.chmod(out, st.st_mode | stat.S_IEXEC)
p = Popen(["/usr/bin/ls"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, env={"LD_LIBRARY_PATH": tmp_dir})
stdout, _ = p.communicate()
self.logger.debug(stdout.decode("utf8"))
self.assertIsNotNone(re.search(r'LIEF is Working', stdout.decode("utf8")))
if os.path.isdir(tmp_dir):
shutil.rmtree(tmp_dir)
def tearDown(self):
# Delete it
if os.path.isdir(self.tmp_dir):
shutil.rmtree(self.tmp_dir)
if __name__ == '__main__':
root_logger = logging.getLogger()
root_logger.setLevel(logging.DEBUG)
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
root_logger.addHandler(ch)
unittest.main(verbosity=2)