diff --git a/poetry.lock b/poetry.lock index 2d99b4731..ec3120ac6 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1,4 +1,4 @@ -# This file is automatically @generated by Poetry 2.2.1 and should not be changed by hand. +# This file is automatically @generated by Poetry 2.4.1 and should not be changed by hand. [[package]] name = "antlr4-python3-runtime" @@ -396,6 +396,93 @@ files = [ {file = "keystone_engine-0.9.2-py2.py3-none-win_amd64.whl", hash = "sha256:c91db1ff16d9d094e00d1827107d1b4afd5e63ce19b491a0140e660635000e8b"}, ] +[[package]] +name = "lief" +version = "1.0.0" +description = "Library to instrument executable formats" +optional = false +python-versions = ">=3.9" +groups = ["main"] +files = [ + {file = "lief-1.0.0-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:cd211a2f11afba7330bc7a51e494a658e8695711912ea00e54efad5cf1cf5afe"}, + {file = "lief-1.0.0-cp310-cp310-macosx_11_0_x86_64.whl", hash = "sha256:e317252e743d7f0020598b9120ecc81ba1c569c68d30d50d99dbd1992036d7a4"}, + {file = "lief-1.0.0-cp310-cp310-manylinux_2_28_aarch64.whl", hash = "sha256:cd7512a23b20c7c47a11bc27925e8c324c062a393e23716d7cee866c4d92db1f"}, + {file = "lief-1.0.0-cp310-cp310-manylinux_2_28_i686.whl", hash = "sha256:b14131d013439d2562d0da09b3c316c7df9e3be25c4a4553a0b324be84785f74"}, + {file = "lief-1.0.0-cp310-cp310-manylinux_2_28_riscv64.whl", hash = "sha256:66d63183afe5dcf6721b28a59be758636d85751d2567ada602f6ce40b6d9c508"}, + {file = "lief-1.0.0-cp310-cp310-manylinux_2_28_x86_64.whl", hash = "sha256:5b40bb3591c2db6aea6208b51ef7cb9a9e05c8be5c9437314857d1fda1e61f21"}, + {file = "lief-1.0.0-cp310-cp310-musllinux_1_2_aarch64.whl", hash = "sha256:b0ba45425a9963eec324002169881f3bfe23b34aab20d90e2370f1550a80b78b"}, + {file = "lief-1.0.0-cp310-cp310-musllinux_1_2_i686.whl", hash = "sha256:ae13bc7797c779a0528c5c4339afca38669860d681f958cc04a9b2e979cdc783"}, + {file = "lief-1.0.0-cp310-cp310-musllinux_1_2_riscv64.whl", hash = "sha256:1d782903e43f71bf9ac7224463a30ef74e975914d463a462288437affe8ad3d2"}, + {file = "lief-1.0.0-cp310-cp310-musllinux_1_2_x86_64.whl", hash = "sha256:64b48a18be73768f6259e4c0fd0ac65202c2eae54a8629be026296027a42c539"}, + {file = "lief-1.0.0-cp310-cp310-win32.whl", hash = "sha256:7e8fde88db83014bce91269188393c41ec81ddd47f32726a1f8f93a5be6be42e"}, + {file = "lief-1.0.0-cp310-cp310-win_amd64.whl", hash = "sha256:b3f173de0e49ad89fc4ea1e74508da3220814a44728060b494d85b8998726760"}, + {file = "lief-1.0.0-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:e9037bda164285b2a59ee3910dfb70e775b2e3ba793edc565725e0bf926b09e9"}, + {file = "lief-1.0.0-cp311-cp311-macosx_11_0_x86_64.whl", hash = "sha256:2edb6c75944a90e1f9b6fb0e086844b7d00311344149f4007dd4f029476f8fe2"}, + {file = "lief-1.0.0-cp311-cp311-manylinux_2_28_aarch64.whl", hash = "sha256:95fb7dc84960068ab881bc681e646cab55f3d736c07bb07e91d5cbc8738885a7"}, + {file = "lief-1.0.0-cp311-cp311-manylinux_2_28_i686.whl", hash = "sha256:ea1220b0b68a9b2cd6d1b3f05276c877d136f5b39e3a01d3433df70a92803071"}, + {file = "lief-1.0.0-cp311-cp311-manylinux_2_28_riscv64.whl", hash = "sha256:5aa92fa9501d60c0dbc8fe88d40fa9f19708fad958674eaea1b52cf4217e734d"}, + {file = "lief-1.0.0-cp311-cp311-manylinux_2_28_x86_64.whl", hash = "sha256:96277e45d58de45a33249079dac43e50db1c96da162fbe1ad02fbaa82e3bd06f"}, + {file = "lief-1.0.0-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:514ba1c9c94f0cd09986bfb1db71fd678acd19f826058fe9cefbba0e7b9a5d86"}, + {file = "lief-1.0.0-cp311-cp311-musllinux_1_2_i686.whl", hash = "sha256:7ed5d9511cd7313bb56c95c4f580e8e332615ae0bd45e05b43990f1b563ceef3"}, + {file = "lief-1.0.0-cp311-cp311-musllinux_1_2_riscv64.whl", hash = "sha256:37810cadad1fd796b6a681fe7eac084489103a7b2fed306776f9932b38c8d4ac"}, + {file = "lief-1.0.0-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:34d12d846529e5d5157a7feaf54d1bd8b6c2905cb6c46cc1f7c94fa9e9bd1f6a"}, + {file = "lief-1.0.0-cp311-cp311-win32.whl", hash = "sha256:1a20cd029b539eb5888882b377617544da012e1bbb2ce575a524c87aaec1102b"}, + {file = "lief-1.0.0-cp311-cp311-win_amd64.whl", hash = "sha256:fbaf64c338086da0a5d67b0e440b7010f2fd4b1d6edc68db303693759320ebfb"}, + {file = "lief-1.0.0-cp311-cp311-win_arm64.whl", hash = "sha256:a9b7bae94a3a160af9b52f15265e55a59a1f3bb305ae9280cfe9ce84cf2c5d0e"}, + {file = "lief-1.0.0-cp312-abi3-macosx_11_0_arm64.whl", hash = "sha256:92bf3b06c19a1ebd9e7978de47cc61040752ec452f7906bc3aac8d57abb23a8b"}, + {file = "lief-1.0.0-cp312-abi3-macosx_11_0_x86_64.whl", hash = "sha256:48e360eea660d019312e377954f327e4fe0918781ee45e9056cd641f50bb6355"}, + {file = "lief-1.0.0-cp312-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:a9e4b9e165bde0163624e159c8c7a096e3d6f671371c70d008baac7ac1f04f0d"}, + {file = "lief-1.0.0-cp312-abi3-manylinux_2_28_i686.whl", hash = "sha256:6b5d975f45e8830bc12da346d3dd317803d8852f1b61838e655741178e46d300"}, + {file = "lief-1.0.0-cp312-abi3-manylinux_2_28_riscv64.whl", hash = "sha256:5ab0c89879066a36467fb8b85485710e40ea982c90f46de3486d6633b1d43e2f"}, + {file = "lief-1.0.0-cp312-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:41ff868f5328fc8e5237a9e1f4718590e0ee9180c1422b3e4948b7116ba4517c"}, + {file = "lief-1.0.0-cp312-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:30c48bb32afdcea98df04cd456e7199e84a629396faf475ca9c8880a4b4c1ce3"}, + {file = "lief-1.0.0-cp312-abi3-musllinux_1_2_i686.whl", hash = "sha256:f58e26e2e4d110cd6a42d86ab497cd4bedc47d72690fae557023ac536d85edcc"}, + {file = "lief-1.0.0-cp312-abi3-musllinux_1_2_riscv64.whl", hash = "sha256:f6c2007bf96704cedc5cee8ae303ba7008840b9a2214ca9f4f43a6faa703834f"}, + {file = "lief-1.0.0-cp312-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:714c96894048f15c8fcecbe6ef4a95416a0f256a20cc6130f8f2398ee717ca06"}, + {file = "lief-1.0.0-cp312-abi3-win32.whl", hash = "sha256:d552235f6c7b999b1837946b61cc64537501fac4fda1663c2cb76482b27b609f"}, + {file = "lief-1.0.0-cp312-abi3-win_amd64.whl", hash = "sha256:ca7774b0d88f4528a1c153d7a9d2798800d6389847b966861fad9f2edb17a593"}, + {file = "lief-1.0.0-cp312-abi3-win_arm64.whl", hash = "sha256:956ee963e2a2ed318d08fd9fe4ff948906c483db1b88f7a62dac24db76ffa4b0"}, + {file = "lief-1.0.0-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:23253bbc154ff0dc5253a8929a2a407d11d459694c409d053b93308e623125ff"}, + {file = "lief-1.0.0-cp314-cp314t-macosx_11_0_x86_64.whl", hash = "sha256:ecd440475e677c6834af996d34a2babbad1af6a3addaaa3822fd021972c41311"}, + {file = "lief-1.0.0-cp314-cp314t-manylinux_2_28_aarch64.whl", hash = "sha256:0e4c0934c27bea29752bdac494c2611c676d27d54139c96098f57b22d065deeb"}, + {file = "lief-1.0.0-cp314-cp314t-manylinux_2_28_i686.whl", hash = "sha256:4d2f1565110abc06070c0f6dcababab64606721fd1401477857558d9575ad663"}, + {file = "lief-1.0.0-cp314-cp314t-manylinux_2_28_riscv64.whl", hash = "sha256:25ea4a80545eb4d1ffe75f49fceca526ce85130bd263bcd81a0fa66178ab1e7f"}, + {file = "lief-1.0.0-cp314-cp314t-manylinux_2_28_x86_64.whl", hash = "sha256:18b05ce0244be151c58fedbd8b08687ed69713533cab9af23060a73f4441e847"}, + {file = "lief-1.0.0-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:26777c1ec1ac55d3b981cad0fdb61d38de21da4b83810d637a5f4d0190470bf6"}, + {file = "lief-1.0.0-cp314-cp314t-musllinux_1_2_i686.whl", hash = "sha256:5881e7698fb9b7ea6398dc86db78c6ae65331d7473c713d5a24318e264048334"}, + {file = "lief-1.0.0-cp314-cp314t-musllinux_1_2_riscv64.whl", hash = "sha256:6be55884009b1eea2314c6d918a945c8bc72584928957771956b600fdcecaa19"}, + {file = "lief-1.0.0-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:f12816bb3d4781fdc98db4df312a8da744c5a93f587fa4919124fcb9b277d510"}, + {file = "lief-1.0.0-cp314-cp314t-win32.whl", hash = "sha256:b5f954b86f6c99437a461ed794bbc2d5e003e6a24e462b49f4978bce228b3801"}, + {file = "lief-1.0.0-cp314-cp314t-win_amd64.whl", hash = "sha256:4680aeaa0eb9f21540f37ab5f97820fdf8f08b9c67c42b4435e232ca8cc22a27"}, + {file = "lief-1.0.0-cp314-cp314t-win_arm64.whl", hash = "sha256:7677a7b8fffd4de096d7ca1db64d1a6c3007e02650ed256fb93b2f3f9ea74097"}, + {file = "lief-1.0.0-cp315-cp315t-macosx_11_0_arm64.whl", hash = "sha256:47c60ba68c3a1f53d872bb004a4e7d2a6f2572372d79a362fa3bd3369f5b8945"}, + {file = "lief-1.0.0-cp315-cp315t-macosx_11_0_x86_64.whl", hash = "sha256:caad33b4fe1eebf8d18acac3a8e2bbfc10f89706f7555e459e6d2588eb8f80fd"}, + {file = "lief-1.0.0-cp315-cp315t-manylinux_2_28_aarch64.whl", hash = "sha256:23f42c7f4ae2df5492eb3a4ed449a2ae16bec107d9baf987bd618fde1ce53564"}, + {file = "lief-1.0.0-cp315-cp315t-manylinux_2_28_i686.whl", hash = "sha256:980aeaff59734b48927b3f9a2915c7d834c16d75d9a36a581e4f0dbd082481a8"}, + {file = "lief-1.0.0-cp315-cp315t-manylinux_2_28_riscv64.whl", hash = "sha256:6a02dcd93c9dba14d3f007bc3a2a84ebf2706b704a5ab32ea6bf71987d4a9f5b"}, + {file = "lief-1.0.0-cp315-cp315t-manylinux_2_28_x86_64.whl", hash = "sha256:fcaf9a28f004bc5f49637dd1a5cdfb8f937feb58febb5f70ebec29c0c1c3584c"}, + {file = "lief-1.0.0-cp315-cp315t-musllinux_1_2_aarch64.whl", hash = "sha256:4e6ae6b3bfd4bc4b2f8a805d0dc74b00de9da173d9a4b2bd581d774bb2cdd19d"}, + {file = "lief-1.0.0-cp315-cp315t-musllinux_1_2_i686.whl", hash = "sha256:2ee72db426d304a37d7bcfe246de3ea1e524ff95104c39f132b88835e6408fa6"}, + {file = "lief-1.0.0-cp315-cp315t-musllinux_1_2_riscv64.whl", hash = "sha256:0628dea883631d33a84498e79c96e9cffff24ed898f9204d139ab1a51e079478"}, + {file = "lief-1.0.0-cp315-cp315t-musllinux_1_2_x86_64.whl", hash = "sha256:5f7f187745c81a81231d9785bed73bae90f495531c5ea40449370e37441b45a2"}, + {file = "lief-1.0.0-cp315-cp315t-win32.whl", hash = "sha256:7a9c376fd607a9480777d0928ae37fd48547937b3a54bb5d130512061e5e953c"}, + {file = "lief-1.0.0-cp315-cp315t-win_amd64.whl", hash = "sha256:4d49b2fa555163692dbecba147a55847abf641fee3c6d72315db14d084b4a160"}, + {file = "lief-1.0.0-cp315-cp315t-win_arm64.whl", hash = "sha256:580e786d72fc34a0402b0af66ed755e668293946705b8193b766c501f00bc448"}, + {file = "lief-1.0.0-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:a4d1c2e764ab609fbc81d140b4949039dc728e83e060a68437b3eb0350121a29"}, + {file = "lief-1.0.0-cp39-cp39-macosx_11_0_x86_64.whl", hash = "sha256:b2d9a1c08ce23112ee1141d9c3e799d03695923c9e383c90edd0b0856b36483d"}, + {file = "lief-1.0.0-cp39-cp39-manylinux_2_28_aarch64.whl", hash = "sha256:9b5be5f2c08a9fbcf3f37f19aa297fb567ec36e939c24ecbe6d45c4a7b3a981b"}, + {file = "lief-1.0.0-cp39-cp39-manylinux_2_28_i686.whl", hash = "sha256:c68c3c19eb75acbca43970fd68a6ccd5fb8795582d31457dcfda744307b54c00"}, + {file = "lief-1.0.0-cp39-cp39-manylinux_2_28_riscv64.whl", hash = "sha256:cc8838a75f93ee9dbd784e6da7ee527d11d16a83de18f8c975344231429da8f8"}, + {file = "lief-1.0.0-cp39-cp39-manylinux_2_28_x86_64.whl", hash = "sha256:9ef6ed0bd319ace01168c4fe5fcf09dbf2245e339f249fad80a2bcebb0439d3d"}, + {file = "lief-1.0.0-cp39-cp39-musllinux_1_2_aarch64.whl", hash = "sha256:bac71c133a9412006a07010eefa6438032db8641aedcc8f6ace8fe845fb97067"}, + {file = "lief-1.0.0-cp39-cp39-musllinux_1_2_i686.whl", hash = "sha256:3cf41cda0c748687e58295117004d44b499005e396a34f58ac8a514bfc899491"}, + {file = "lief-1.0.0-cp39-cp39-musllinux_1_2_riscv64.whl", hash = "sha256:4e3abaac54cb548b7e8d279f3008e990ee5184eea0fa85524056a9522e36ed54"}, + {file = "lief-1.0.0-cp39-cp39-musllinux_1_2_x86_64.whl", hash = "sha256:26f2d29dadd36b850f38f58ecca97eebbd3bf201a1ec13355a31177b44f4cc50"}, + {file = "lief-1.0.0-cp39-cp39-win32.whl", hash = "sha256:ca999ccd12aba1c09d4d1337624c1fbc3a4084c1714f12a5be8bb9d1484f0d42"}, + {file = "lief-1.0.0-cp39-cp39-win_amd64.whl", hash = "sha256:3fd100c594bd5451d4c22b67e42d286860869ebb56d98f635c1daa56a11760e1"}, + {file = "lief-1.0.0.tar.gz", hash = "sha256:811b2354f48fa08c49b106eafc9c1b26006371dc24d3d04bfdd9939863df6e6a"}, +] + [[package]] name = "loguru" version = "0.7.2" @@ -413,7 +500,7 @@ colorama = {version = ">=0.3.4", markers = "sys_platform == \"win32\""} win32-setctime = {version = ">=1.0.0", markers = "sys_platform == \"win32\""} [package.extras] -dev = ["Sphinx (==7.2.5) ; python_version >= \"3.9\"", "colorama (==0.4.5) ; python_version < \"3.8\"", "colorama (==0.4.6) ; python_version >= \"3.8\"", "exceptiongroup (==1.1.3) ; python_version >= \"3.7\" and python_version < \"3.11\"", "freezegun (==1.1.0) ; python_version < \"3.8\"", "freezegun (==1.2.2) ; python_version >= \"3.8\"", "mypy (==v0.910) ; python_version < \"3.6\"", "mypy (==v0.971) ; python_version == \"3.6\"", "mypy (==v1.4.1) ; python_version == \"3.7\"", "mypy (==v1.5.1) ; python_version >= \"3.8\"", "pre-commit (==3.4.0) ; python_version >= \"3.8\"", "pytest (==6.1.2) ; python_version < \"3.8\"", "pytest (==7.4.0) ; python_version >= \"3.8\"", "pytest-cov (==2.12.1) ; python_version < \"3.8\"", "pytest-cov (==4.1.0) ; python_version >= \"3.8\"", "pytest-mypy-plugins (==1.9.3) ; python_version >= \"3.6\" and python_version < \"3.8\"", "pytest-mypy-plugins (==3.0.0) ; python_version >= \"3.8\"", "sphinx-autobuild (==2021.3.14) ; python_version >= \"3.9\"", "sphinx-rtd-theme (==1.3.0) ; python_version >= \"3.9\"", "tox (==3.27.1) ; python_version < \"3.8\"", "tox (==4.11.0) ; python_version >= \"3.8\""] +dev = ["Sphinx (==7.2.5) ; python_version >= \"3.9\"", "colorama (==0.4.5) ; python_version < \"3.8\"", "colorama (==0.4.6) ; python_version >= \"3.8\"", "exceptiongroup (==1.1.3) ; python_version >= \"3.7\" and python_version < \"3.11\"", "freezegun (==1.1.0) ; python_version < \"3.8\"", "freezegun (==1.2.2) ; python_version >= \"3.8\"", "mypy (==0.910) ; python_version < \"3.6\"", "mypy (==0.971) ; python_version == \"3.6\"", "mypy (==1.4.1) ; python_version == \"3.7\"", "mypy (==1.5.1) ; python_version >= \"3.8\"", "pre-commit (==3.4.0) ; python_version >= \"3.8\"", "pytest (==6.1.2) ; python_version < \"3.8\"", "pytest (==7.4.0) ; python_version >= \"3.8\"", "pytest-cov (==2.12.1) ; python_version < \"3.8\"", "pytest-cov (==4.1.0) ; python_version >= \"3.8\"", "pytest-mypy-plugins (==1.9.3) ; python_version >= \"3.6\" and python_version < \"3.8\"", "pytest-mypy-plugins (==3.0.0) ; python_version >= \"3.8\"", "sphinx-autobuild (==2021.3.14) ; python_version >= \"3.9\"", "sphinx-rtd-theme (==1.3.0) ; python_version >= \"3.9\"", "tox (==3.27.1) ; python_version < \"3.8\"", "tox (==4.11.0) ; python_version >= \"3.8\""] [[package]] name = "multiprocess" @@ -1152,4 +1239,4 @@ re = ["r2libr"] [metadata] lock-version = "2.1" python-versions = "^3.10" -content-hash = "b65631e55580e06f60d535d41d2c2d0c4ee786609c7232988ceda8ce6b586a6d" +content-hash = "c95d287016c7cf976cd461a38d7e287c7dd6d5e9394159f22c014a007d68aa1f" diff --git a/pyproject.toml b/pyproject.toml index afa35a8d2..a3ea208f5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -44,6 +44,7 @@ unicornafl = { version = "^2.0.0", markers = "platform_system != 'Windows'", opt fuzzercorn = { version = "^0.0.1", markers = "platform_system == 'Linux'", optional = true } r2libr = { version = "^5.7.4", optional = true } +lief = "^1.0.0" [tool.poetry.extras] fuzz = ["unicornafl", "fuzzercorn"] diff --git a/qiling/loader/elf.py b/qiling/loader/elf.py index a5a8ac6fe..471d8f22b 100644 --- a/qiling/loader/elf.py +++ b/qiling/loader/elf.py @@ -9,12 +9,7 @@ from enum import IntEnum from typing import Any, AnyStr, Optional, Sequence, Mapping, Tuple -from elftools.common.utils import preserve_stream_pos -from elftools.elf.constants import P_FLAGS, SH_FLAGS -from elftools.elf.elffile import ELFFile -from elftools.elf.relocation import RelocationHandler -from elftools.elf.sections import Symbol, SymbolTableSection -from elftools.elf.descriptions import describe_reloc_type +from lief import ELF from unicorn.unicorn_const import UC_PROT_NONE, UC_PROT_READ, UC_PROT_WRITE, UC_PROT_EXEC from qiling import Qiling @@ -65,6 +60,28 @@ class AUXV(IntEnum): SYSCALL_SIZE = 0x1000 +def _iter_lief_relocations(binary: ELF.Binary): + """Primary relocation iterator using LIEF's parsed relocations. + + Yields (target_sec, r_sym, r_offset, r_type, r_addend) with r_type as + the ELF.Relocation.TYPE enum value for direct comparison. + """ + cls = binary.header.identity_class + is_64 = cls == ELF.Header.CLASS.ELF64 + + for rel in binary.relocations: + target_sec = rel.section + if target_sec is None: + continue + + r_info = rel.r_info(cls) + r_sym = r_info >> 32 if is_64 else r_info >> 8 + r_offset = rel.address + r_addend = rel.addend + + yield target_sec, r_sym, r_offset, rel.type, r_addend + + class QlLoaderELF(QlLoader): def __init__(self, ql: Qiling): super().__init__(ql) @@ -93,32 +110,36 @@ def run(self): self.path = self.ql.path with open(self.path, 'rb') as infile: - fstream = io.BytesIO(infile.read()) + raw = infile.read() - elffile = ELFFile(fstream) - elftype = elffile['e_type'] + binary = ELF.parse(list(raw)) + + if binary is None: + raise QlErrorELFFormat('failed to parse ELF file') + + elftype = binary.header.file_type # is it a driver? - if elftype == 'ET_REL': - self.load_driver(elffile, top_of_stack, loadbase=0x8000000) + if elftype == ELF.Header.FILE_TYPE.REL: + self.load_driver(binary, raw, top_of_stack, loadbase=0x8000000) self.ql.hook_code(hook_kernel_api) # is it an executable? - elif elftype == 'ET_EXEC': + elif elftype == ELF.Header.FILE_TYPE.EXEC: load_address = 0 - self.load_with_ld(elffile, top_of_stack, load_address, self.argv, self.env) + self.load_with_ld(binary, top_of_stack, load_address, self.argv, self.env) # is it a shared object? - elif elftype == 'ET_DYN': + elif elftype == ELF.Header.FILE_TYPE.DYN: load_address = self.profile.getint('load_address') - self.load_with_ld(elffile, top_of_stack, load_address, self.argv, self.env) + self.load_with_ld(binary, top_of_stack, load_address, self.argv, self.env) else: raise QlErrorELFFormat(f'unexpected elf type value (e_type = {elftype})') - self.is_driver = (elftype == 'ET_REL') + self.is_driver = (elftype == ELF.Header.FILE_TYPE.REL) self.ql.arch.regs.arch_sp = self.stack_address @@ -135,22 +156,25 @@ def seg_perm_to_uc_prot(perm: int) -> int: prot = UC_PROT_NONE - if perm & P_FLAGS.PF_X: + if perm & 0x1: # PF_X prot |= UC_PROT_EXEC - if perm & P_FLAGS.PF_W: + if perm & 0x2: # PF_W prot |= UC_PROT_WRITE - if perm & P_FLAGS.PF_R: + if perm & 0x4: # PF_R prot |= UC_PROT_READ return prot - def load_with_ld(self, elffile: ELFFile, stack_addr: int, load_address: int, argv: Sequence[str] = [], env: Mapping[AnyStr, AnyStr] = {}): + def load_with_ld(self, binary: ELF.Binary, stack_addr: int, load_address: int, argv: Sequence[str] = [], env: Mapping[str, str] = {}): - def load_elf_segments(elffile: ELFFile, load_address: int, info: str): + def load_elf_segments(binary: ELF.Binary, load_address: int, info: str): # get list of loadable segments; these segments will be loaded to memory - load_segments = sorted(elffile.iter_segments(type='PT_LOAD'), key=lambda s: s['p_vaddr']) + load_segments = sorted( + [s for s in binary.segments if s.type == ELF.Segment.TYPE.LOAD], + key=lambda s: s.virtual_address + ) # determine the memory regions that need to be mapped in order to load the segments. # note that region boundaries are aligned to page, which means they may be larger than @@ -160,9 +184,9 @@ def load_elf_segments(elffile: ELFFile, load_address: int, info: str): # iterate over loadable segments for seg in load_segments: - lbound = self.ql.mem.align(load_address + seg['p_vaddr']) - ubound = self.ql.mem.align_up(load_address + seg['p_vaddr'] + seg['p_memsz']) - perms = QlLoaderELF.seg_perm_to_uc_prot(seg['p_flags']) + lbound = self.ql.mem.align(load_address + seg.virtual_address) + ubound = self.ql.mem.align_up(load_address + seg.virtual_address + seg.virtual_size) + perms = QlLoaderELF.seg_perm_to_uc_prot(int(seg.flags)) if load_regions: prev_lbound, prev_ubound, prev_perms = load_regions[-1] @@ -213,12 +237,12 @@ def load_elf_segments(elffile: ELFFile, load_address: int, info: str): # load loadable segments contents to memory for seg in load_segments: - self.ql.mem.write(load_address + seg['p_vaddr'], seg.data()) + self.ql.mem.write(load_address + seg.virtual_address, bytes(seg.content)) return load_regions[0][0], load_regions[-1][1] - mem_start, mem_end = load_elf_segments(elffile, load_address, self.path) - self.elf_entry = entry_point = load_address + elffile['e_entry'] + mem_start, mem_end = load_elf_segments(binary, load_address, self.path) + self.elf_entry = entry_point = load_address + binary.header.entrypoint self.ql.log.debug(f'mem_start : {mem_start:#x}') self.ql.log.debug(f'mem_end : {mem_end:#x}') @@ -230,8 +254,7 @@ def load_elf_segments(elffile: ELFFile, load_address: int, info: str): self.brk_address = mem_end + 0x2000 # determine interpreter path - interp_seg = next(elffile.iter_segments(type='PT_INTERP'), None) - interp_path = str(interp_seg.get_interp_name()) if interp_seg else '' + interp_path = binary.interpreter # '' if no PT_INTERP segment interp_address = 0 @@ -245,24 +268,27 @@ def load_elf_segments(elffile: ELFFile, load_address: int, info: str): if not self.ql.os.path.is_safe_host_path(interp_hpath): raise PermissionError(f'unsafe path: {interp_hpath}') - with open(interp_hpath, 'rb') as infile: - interp = ELFFile(infile) - min_vaddr = min(seg['p_vaddr'] for seg in interp.iter_segments(type='PT_LOAD')) + interp_binary = ELF.parse(interp_hpath) + if interp_binary is None: + raise QlErrorELFFormat(f'failed to parse interpreter: {interp_hpath}') - # determine interpreter base address - # some old interpreters may not be PIE: p_vaddr of the first LOAD segment is not zero - # we should load interpreter at the address p_vaddr specified in such situation - interp_address = self.profile.getint('interp_address') if min_vaddr == 0 else 0 - self.ql.log.debug(f'Interpreter addr: {interp_address:#x}') + interp_load_segs = [s for s in interp_binary.segments if s.type == ELF.Segment.TYPE.LOAD] + min_vaddr = min(s.virtual_address for s in interp_load_segs) - # load interpreter segments data to memory - interp_start, interp_end = load_elf_segments(interp, interp_address, interp_vpath) + # determine interpreter base address + # some old interpreters may not be PIE: p_vaddr of the first LOAD segment is not zero + # we should load interpreter at the address p_vaddr specified in such situation + interp_address = self.profile.getint('interp_address') if min_vaddr == 0 else 0 + self.ql.log.debug(f'Interpreter addr: {interp_address:#x}') - # add interpreter to the loaded images list - self.images.append(Image(interp_start, interp_end, interp_hpath)) + # load interpreter segments data to memory + interp_start, interp_end = load_elf_segments(interp_binary, interp_address, interp_vpath) - # determine entry point - entry_point = interp_address + interp['e_entry'] + # add interpreter to the loaded images list + self.images.append(Image(interp_start, interp_end, interp_hpath)) + + # determine entry point + entry_point = interp_address + interp_binary.header.entrypoint # set mmap addr mmap_address = self.profile.getint('mmap_address') @@ -321,9 +347,9 @@ def __push_str(top: int, s: str) -> int: new_stack = execfn = __push_str(new_stack, argv[0]) # store aux vector data for gdb use - elf_phdr = elffile['e_phoff'] + mem_start - elf_phent = elffile['e_phentsize'] - elf_phnum = elffile['e_phnum'] + elf_phdr = binary.header.program_header_offset + mem_start + elf_phent = binary.header.program_header_size + elf_phnum = binary.header.numberof_segments # for more details on the following values see: # https://github.com/google/cpu_features/blob/main/include/internal/hwcaps.h @@ -420,29 +446,23 @@ def __assemble(asm: str) -> bytes: self.ql.mem.write(vsyscall_addr + i * entry_size, entry.ljust(entry_size, b'\xcc')) - def lkm_get_init(self, elffile: ELFFile) -> int: + def lkm_get_init(self, binary: ELF.Binary) -> int: """Get file offset of the init_module function. """ - symbol_tables = (sec for sec in elffile.iter_sections() if type(sec) is SymbolTableSection) - - for sec in symbol_tables: - syms = sec.get_symbol_by_name('init_module') + sym = binary.get_symbol('init_module') - if syms: - sym = syms[0] - addr = sym['st_value'] + elffile.get_section(sym['st_shndx'])['sh_offset'] - - return addr + if sym is not None: + return sym.value + binary.sections[sym.shndx].offset raise QlErrorELFFormat('invalid module: symbol init_module not found') - def lkm_dynlinker(self, elffile: ELFFile, mem_start: int) -> Mapping[str, int]: - def __get_symbol(name: str) -> Optional[Symbol]: - _symtab = elffile.get_section_by_name('.symtab') - _sym = _symtab.get_symbol_by_name(name) + def lkm_dynlinker(self, binary: ELF.Binary, raw: bytes, mem_start: int) -> Mapping[str, int]: + # Index symbols by name for fast lookup + sym_by_name = {sym.name: sym for sym in binary.symbols if sym.name} - return _sym[0] if _sym else None + def __get_symbol(name: str): + return sym_by_name.get(name) ql = self.ql @@ -453,148 +473,166 @@ def __get_symbol(name: str) -> Optional[Symbol]: # reverse dictionary to map symbol name -> address rev_reloc_symbols = {} - rh = RelocationHandler(elffile) - sections = [sec for sec in elffile.iter_sections() if sec['sh_flags'] & SH_FLAGS.SHF_ALLOC] + # Build a list of all ELF symbols for index-based lookup (for anonymous symbols) + all_elf_symbols = list(binary.symbols) + sections_list = list(binary.sections) + + SHF_ALLOC = int(ELF.Section.FLAGS.ALLOC) + alloc_section_names = {sec.name for sec in sections_list if sec.flags & SHF_ALLOC} + + # binary.header.machine_type is correct even for big-endian targets — + # LIEF handles ELF header byte-swapping correctly for all architectures. + e_machine = binary.header.machine_type + + T = ELF.Relocation.TYPE + A = ELF.ARCH + + prev_mips_hi16_loc = 0 # used by R_MIPS_HI16/LO16 pair + + reloc_source = _iter_lief_relocations(binary) - for sec in sections: - reloc_sec = rh.find_relocations_for_section(sec) + for target_sec, r_sym, r_offset, r_type, r_addend in reloc_source: + # skip relocations for non-alloc sections (e.g. .gnu.linkonce.this_module) + if target_sec.name not in alloc_section_names: + continue + + # Look up symbol by index + symbol = all_elf_symbols[r_sym] if (0 < r_sym < len(all_elf_symbols)) else None + + # sym_offset defaults to the target section offset (for named symbols) + sym_offset = target_sec.offset + + if symbol is not None and symbol.name == '': + # SECTION-type anonymous symbol: resolve via symbol.shndx to the actual referenced section + if 0 < symbol.shndx < len(sections_list): + symsec = sections_list[symbol.shndx] + symbol_name = symsec.name + sym_offset = symsec.offset + rev_reloc_symbols[symbol_name] = sym_offset + mem_start + else: + continue + elif symbol is None: + # r_sym == 0: null symbol, section-relative to target section itself + symbol_name = target_sec.name + rev_reloc_symbols[symbol_name] = sym_offset + mem_start + else: + symbol_name = symbol.name + + if symbol_name in all_symbols: + sym_offset = rev_reloc_symbols[symbol_name] - mem_start + else: + all_symbols.append(symbol_name) + _symbol = __get_symbol(symbol_name) - if reloc_sec and reloc_sec.name != '.rela.gnu.linkonce.this_module': - # get the symbol table section pointed in sh_link - symtab = elffile.get_section(reloc_sec['sh_link']) - assert isinstance(symtab, SymbolTableSection) + if _symbol is None or _symbol.shndx == 0: # SHN_UNDEF + # external symbol + # only save symbols of APIs - for rel in reloc_sec.iter_relocations(): - # if reloc['r_info_sym'] == 0: - # continue + # we need to lookup from address to symbol, so we can find the right callback + # for sys_xxx handler for syscall, the address must be aligned to pointer size + if symbol_name.startswith('sys_'): + self.ql.os.hook_addr = self.ql.mem.align_up(self.ql.os.hook_addr, self.ql.arch.pointersize) - symbol = symtab.get_symbol(rel['r_info_sym']) - assert symbol + self.import_symbols[self.ql.os.hook_addr] = symbol_name - # Some symbols have zero 'st_name', so instead what's used is - # the name of the section they point at. - if symbol['st_name'] == 0: - symsec = elffile.get_section(symbol['st_shndx']) - symbol_name = symsec.name - sym_offset = symsec['sh_offset'] + # FIXME: this is for rootkit to scan for syscall table from page_offset_base + # write address of syscall table to this slot, so syscall scanner can quickly find it + if symbol_name == "page_offset_base": + ql.mem.write_ptr(self.ql.os.hook_addr, SYSCALL_MEM) + + # we also need to do reverse lookup from symbol to address + rev_reloc_symbols[symbol_name] = self.ql.os.hook_addr + sym_offset = self.ql.os.hook_addr - mem_start + self.ql.os.hook_addr += self.ql.arch.pointersize + + elif _symbol.shndx == 0xfff1: # SHN_ABS + rev_reloc_symbols[symbol_name] = _symbol.value - rev_reloc_symbols[symbol_name] = sym_offset + mem_start else: - symbol_name = symbol.name - # get info about related section to be patched - info_section = elffile.get_section(reloc_sec['sh_info']) - sym_offset = info_section['sh_offset'] + # local symbol + _section = list(binary.sections)[_symbol.shndx] + rev_reloc_symbols[symbol_name] = _section.offset + _symbol.value + mem_start - if symbol_name in all_symbols: - sym_offset = rev_reloc_symbols[symbol_name] - mem_start - else: - all_symbols.append(symbol_name) - _symbol = __get_symbol(symbol_name) + # ql.log.info(f'relocating: {symbol_name} -> {rev_reloc_symbols[symbol_name]:#010x}') - if _symbol['st_shndx'] == 'SHN_UNDEF': - # external symbol - # only save symbols of APIs + loc = target_sec.offset + r_offset + mem_start - # we need to lookup from address to symbol, so we can find the right callback - # for sys_xxx handler for syscall, the address must be aligned to pointer size - if symbol_name.startswith('sys_'): - self.ql.os.hook_addr = self.ql.mem.align_up(self.ql.os.hook_addr, self.ql.arch.pointersize) + if e_machine == A.X86_64: + if r_type in (T.X86_64_32S, T.X86_64_32): + if r_addend: + val = sym_offset + r_addend + mem_start + else: + val = rev_reloc_symbols[symbol_name] + ql.mem.write_ptr(loc, (val & 0xFFFFFFFF), 4) - self.import_symbols[self.ql.os.hook_addr] = symbol_name + elif r_type == T.X86_64_64: + val = sym_offset + r_addend + 0x2000000 # init_module position: FIXME + ql.mem.write_ptr(loc, val, 8) - # FIXME: this is for rootkit to scan for syscall table from page_offset_base - # write address of syscall table to this slot, so syscall scanner can quickly find it - if symbol_name == "page_offset_base": - ql.mem.write_ptr(self.ql.os.hook_addr, SYSCALL_MEM) + elif r_type == T.X86_64_PC64: + val = r_addend - loc + rev_reloc_symbols[symbol_name] + ql.mem.write_ptr(loc, val, 8) - # we also need to do reverse lookup from symbol to address - rev_reloc_symbols[symbol_name] = self.ql.os.hook_addr - sym_offset = self.ql.os.hook_addr - mem_start - self.ql.os.hook_addr += self.ql.arch.pointersize + elif r_type in (T.X86_64_PC32, T.X86_64_PLT32): + val = r_addend - loc + rev_reloc_symbols[symbol_name] + ql.mem.write_ptr(loc, (val & 0xFFFFFFFF), 4) - elif _symbol['st_shndx'] == 'SHN_ABS': - rev_reloc_symbols[symbol_name] = _symbol['st_value'] + else: + raise NotImplementedError(f'Relocation type {r_type} not implemented for x86_64') - else: - # local symbol - _section = elffile.get_section(_symbol['st_shndx']) - rev_reloc_symbols[symbol_name] = _section['sh_offset'] + _symbol['st_value'] + mem_start + elif e_machine == A.I386: + if r_type in (T.X86_PC32, T.X86_PLT32): + val = ql.mem.read_ptr(loc, 4) + val += rev_reloc_symbols[symbol_name] - loc + ql.mem.write_ptr(loc, (val & 0xFFFFFFFF), 4) - # ql.log.info(f'relocating: {symbol_name} -> {rev_reloc_symbols[symbol_name]:#010x}') + elif r_type == T.X86_32: + val = ql.mem.read_ptr(loc, 4) + val += rev_reloc_symbols[symbol_name] + ql.mem.write_ptr(loc, (val & 0xFFFFFFFF), 4) - # FIXME: using the rh.apply_section_relocations method for the following relocation work - # seems to be cleaner. + else: + raise NotImplementedError(f'Relocation type {r_type} not implemented for i386') - loc = elffile.get_section(reloc_sec['sh_info'])['sh_offset'] + rel['r_offset'] - loc += mem_start + elif e_machine == A.MIPS: + if r_type == T.MIPS_32: + val = ql.mem.read_ptr(loc, 4) + val += rev_reloc_symbols[symbol_name] + ql.mem.write_ptr(loc, (val & 0xFFFFFFFF), 4) - desc = describe_reloc_type(rel['r_info_type'], elffile) + elif r_type == T.MIPS_HI16: + prev_mips_hi16_loc = loc - if desc in ('R_X86_64_32S', 'R_X86_64_32'): - # patch this reloc - if rel['r_addend']: - val = sym_offset + rel['r_addend'] - val += mem_start - else: - val = rev_reloc_symbols[symbol_name] - - ql.mem.write_ptr(loc, (val & 0xFFFFFFFF), 4) - - elif desc == 'R_X86_64_64': - val = sym_offset + rel['r_addend'] - val += 0x2000000 # init_module position: FIXME - ql.mem.write_ptr(loc, val, 8) - - elif desc == 'R_X86_64_PC64': - val = rel['r_addend'] - loc - val += rev_reloc_symbols[symbol_name] - ql.mem.write_ptr(loc, val, 8) - - elif desc in ('R_X86_64_PC32', 'R_X86_64_PLT32'): - val = rel['r_addend'] - loc - val += rev_reloc_symbols[symbol_name] - ql.mem.write_ptr(loc, (val & 0xFFFFFFFF), 4) - - elif desc in ('R_386_PC32', 'R_386_PLT32'): - val = ql.mem.read_ptr(loc, 4) - val += rev_reloc_symbols[symbol_name] - loc - ql.mem.write_ptr(loc, (val & 0xFFFFFFFF), 4) - - elif desc in ('R_386_32', 'R_MIPS_32'): - val = ql.mem.read_ptr(loc, 4) - val += rev_reloc_symbols[symbol_name] - ql.mem.write_ptr(loc, (val & 0xFFFFFFFF), 4) - - elif desc == 'R_MIPS_HI16': - # actual relocation is done in R_MIPS_LO16 - prev_mips_hi16_loc = loc - - elif desc == 'R_MIPS_LO16': - val = ql.mem.read_ptr(prev_mips_hi16_loc + 2, 2) << 16 | ql.mem.read_ptr(loc + 2, 2) - val = rev_reloc_symbols[symbol_name] + val - # *(word)(mips_lo16_loc + 2) is treated as signed - if (val & 0xFFFF) >= 0x8000: - val += (1 << 16) - - ql.mem.write_ptr(prev_mips_hi16_loc + 2, (val >> 16), 2) - ql.mem.write_ptr(loc + 2, (val & 0xFFFF), 2) - - elif desc in ('R_ARM_CALL', 'R_ARM_JUMP24'): - val = (rev_reloc_symbols[symbol_name] - loc - 8) >> 2 - val = (val & 0xFFFFFF) | (ql.mem.read_ptr(loc, 4) & 0xFF000000) - ql.mem.write_ptr(loc, (val & 0xFFFFFFFF), 4) - - elif desc == 'R_ARM_ABS32': - val = rev_reloc_symbols[symbol_name] + ql.mem.read_ptr(loc, 4) - ql.mem.write_ptr(loc, (val & 0xFFFFFFFF), 4) + elif r_type == T.MIPS_LO16: + val = ql.mem.read_ptr(prev_mips_hi16_loc + 2, 2) << 16 | ql.mem.read_ptr(loc + 2, 2) + val = rev_reloc_symbols[symbol_name] + val + # *(word)(mips_lo16_loc + 2) is treated as signed + if (val & 0xFFFF) >= 0x8000: + val += (1 << 16) - else: - raise NotImplementedError(f'Relocation type {desc} not implemented') + ql.mem.write_ptr(prev_mips_hi16_loc + 2, (val >> 16), 2) + ql.mem.write_ptr(loc + 2, (val & 0xFFFF), 2) + + else: + raise NotImplementedError(f'Relocation type {r_type} not implemented for MIPS') + + elif e_machine == A.ARM: + if r_type in (T.ARM_CALL, T.ARM_JUMP24): + val = (rev_reloc_symbols[symbol_name] - loc - 8) >> 2 + val = (val & 0xFFFFFF) | (ql.mem.read_ptr(loc, 4) & 0xFF000000) + ql.mem.write_ptr(loc, (val & 0xFFFFFFFF), 4) + + elif r_type == T.ARM_ABS32: + val = rev_reloc_symbols[symbol_name] + ql.mem.read_ptr(loc, 4) + ql.mem.write_ptr(loc, (val & 0xFFFFFFFF), 4) + + else: + raise NotImplementedError(f'Relocation type {r_type} not implemented for ARM') return rev_reloc_symbols - def load_driver(self, elffile: ELFFile, stack_addr: int, loadbase: int = 0) -> None: - elfdata_mapping = self.get_elfdata_mapping(elffile) + def load_driver(self, binary: ELF.Binary, raw: bytes, stack_addr: int, loadbase: int = 0) -> None: + elfdata_mapping = self.get_elfdata_mapping(binary, raw) mem_start = self.ql.mem.align(loadbase) mem_end = self.ql.mem.align_up(loadbase + len(elfdata_mapping)) @@ -606,11 +644,11 @@ def load_driver(self, elffile: ELFFile, stack_addr: int, loadbase: int = 0) -> N self.ql.log.debug(f'mem_end : {mem_end:#x}') self.ql.mem.map(mem_start, mem_end - mem_start, info=os.path.basename(self.ql.path)) - self.ql.mem.write(loadbase, elfdata_mapping) + self.ql.mem.write(mem_start, elfdata_mapping) self.images.append(Image(mem_start, mem_end, os.path.abspath(self.path))) - init_module = loadbase + self.lkm_get_init(elffile) + init_module = mem_start + self.lkm_get_init(binary) self.ql.log.debug(f'init_module : {init_module:#x}') self.brk_address = mem_end @@ -624,13 +662,13 @@ def load_driver(self, elffile: ELFFile, stack_addr: int, loadbase: int = 0) -> N self.ql.os.entry_point = self.entry_point self.stack_address = self.ql.mem.align(stack_addr, self.ql.arch.pointersize) - self.load_address = loadbase + self.load_address = mem_start # setup syscall table self.ql.mem.map(SYSCALL_MEM, SYSCALL_SIZE, info="[syscall_mem]") self.ql.mem.write(SYSCALL_MEM, b'\x00' * SYSCALL_SIZE) - rev_reloc_symbols = self.lkm_dynlinker(elffile, loadbase) + rev_reloc_symbols = self.lkm_dynlinker(binary, raw, mem_start) # iterate over relocatable symbols, but pick only those who start with 'sys_' for sc, addr in rev_reloc_symbols.items(): @@ -654,33 +692,15 @@ def load_driver(self, elffile: ELFFile, stack_addr: int, loadbase: int = 0) -> N self.import_symbols[self.ql.os.hook_addr + 1 * self.ql.arch.pointersize] = hook_sys_write self.import_symbols[self.ql.os.hook_addr + 2 * self.ql.arch.pointersize] = hook_sys_open - def get_elfdata_mapping(self, elffile: ELFFile) -> bytes: - # from io import BytesIO - # - # rh = RelocationHandler(elffile) - # - # for sec in elffile.iter_sections(): - # rs = rh.find_relocations_for_section(sec) - # - # if rs is not None: - # ss = BytesIO(sec.data()) - # rh.apply_section_relocations(ss, rs) - # - # # apply changes to stream - # elffile.stream.seek(sec['sh_offset']) - # elffile.stream.write(ss.getbuffer()) - # + def get_elfdata_mapping(self, binary: ELF.Binary, raw: bytes) -> bytes: # TODO: need to patch hooked symbols with their hook targets # (e.g. replace calls to 'printk' with the hooked address that # was allocate for it) elfdata_mapping = bytearray() - # pick up elf header - with preserve_stream_pos(elffile.stream): - elffile.stream.seek(0) - elf_header = elffile.stream.read(elffile['e_ehsize']) - + # pick up elf header from raw bytes + elf_header = raw[:binary.header.header_size] elfdata_mapping.extend(elf_header) # FIXME: normally the address of a section would be determined by its 'sh_addr' value. @@ -691,14 +711,16 @@ def get_elfdata_mapping(self, elffile: ELFFile) -> bytes: # here we presume this a relocatable object and don't do any relocation (that is, it # is relocated to 0) + SHF_ALLOC = int(ELF.Section.FLAGS.ALLOC) + # pick up loadable sections - for sec in elffile.iter_sections(): - if sec['sh_flags'] & SH_FLAGS.SHF_ALLOC: + for sec in binary.sections: + if sec.flags & SHF_ALLOC: # pad aggregated elf data to the offset of the current section - elfdata_mapping.extend(b'\x00' * (sec['sh_offset'] - len(elfdata_mapping))) + elfdata_mapping.extend(b'\x00' * (sec.offset - len(elfdata_mapping))) # aggregate section data - elfdata_mapping.extend(sec.data()) + elfdata_mapping.extend(bytes(sec.content)) return bytes(elfdata_mapping) diff --git a/qiling/loader/macho.py b/qiling/loader/macho.py index a7e7f87bc..d84d4c189 100644 --- a/qiling/loader/macho.py +++ b/qiling/loader/macho.py @@ -5,17 +5,14 @@ import os, plistlib, struct +from lief import MachO + from .loader import QlLoader from qiling.exception import * from qiling.const import * - -from .macho_parser.parser import * -from .macho_parser.const import * -from .macho_parser.utils import * from qiling.os.macos.kernel_api.hook import * from qiling.os.memory import QlMemoryHeap - from qiling.os.macos.const import * from qiling.os.macos.task import MachoTask from qiling.os.macos.kernel_func import FileSystem, map_commpage @@ -25,6 +22,40 @@ from qiling.os.macos.thread import QlMachoThreadManagement, QlMachoThread +def _lief_macho_parse(path: str, arch_type) -> 'MachO.Binary': + """Parse a Mach-O file with LIEF, selecting the correct arch slice from a FAT binary.""" + cpu_map = { + QL_ARCH.X8664: MachO.Header.CPU_TYPE.X86_64, + QL_ARCH.ARM64: MachO.Header.CPU_TYPE.ARM64, + } + target_cpu = cpu_map.get(arch_type) + parsed = MachO.parse(path) # always FatBinary, or None on failure + if parsed is None: + raise QlErrorMACHOFormat(f'Failed to parse Mach-O: {path}') + return next( + (b for b in parsed if target_cpu is None or b.header.cpu_type == target_cpu), + parsed[0] + ) + + +def _parse_macho_dysymtab_relocs(raw: bytes, offset: int, count: int) -> list: + """Parse Mach-O scatter relocation entries from raw binary bytes.""" + relocs = [] + for i in range(count): + entry = raw[offset + i*8 : offset + (i+1)*8] + r_address = struct.unpack_from('> 24) & 1, + 'length' : (r_info >> 25) & 3, # 0=1B, 1=2B, 2=4B, 3=8B + 'extern' : (r_info >> 27) & 1, + 'rtype' : (r_info >> 28) & 0xF, + }) + return relocs + + # commpage is a shared mem space which is in a static address def load_commpage(ql): if ql.arch.type == QL_ARCH.X8664: @@ -128,9 +159,13 @@ def run(self): self.ql.os.thread_management.cur_thread = self.ql.os.macho_thread self.ql.os.macho_vmmap_end = vmmap_trap_address self.stack_sp = stack_address + stack_size - self.macho_file = MachoParser(self.ql, self.ql.path) - self.is_driver = (self.macho_file.header.file_type == 0xb) - self.loading_file = self.macho_file + self.macho_file = _lief_macho_parse(self.ql.path, self.ql.arch.type) + self._page_zero_size = next( + (seg.virtual_size for seg in self.macho_file.segments + if seg.virtual_address == 0 and seg.file_size == 0), + 0 + ) + self.is_driver = (self.macho_file.header.file_type == MachO.Header.FILE_TYPE.KEXT_BUNDLE) self.slide = int(self.profile.get("LOADER", "slide"), 16) self.dyld_slide = int(self.profile.get("LOADER", "dyld_slide"), 16) self.string_align = 8 @@ -159,16 +194,15 @@ def loadDriver(self, stack_addr, loadbase = -1, argv = [], env = {}): loadbase = 0xffffff7000000000 self.slide = loadbase self.load_address = loadbase - cmds = self.macho_file.commands - for cmd in cmds: - if cmd.cmd_id == LC_SEGMENT_64: - self.loadSegment64(cmd, False) + for seg in self.macho_file.segments: + self.loadSegment64(seg, False) self.kext_size = self.vm_end_addr - loadbase kernel_path = os.path.join(self.ql.rootfs, "System/Library/Kernels/kernel.development") self.ql.log.info("Parsing kernel:") - self.kernel = MachoParser(self.ql, kernel_path) + self.kernel = _lief_macho_parse(kernel_path, self.ql.arch.type) + self._kernel_raw = open(kernel_path, 'rb').read() # Create memory for external static symbol jmp code self.static_addr = self.vm_end_addr @@ -180,43 +214,56 @@ def loadDriver(self, stack_addr, loadbase = -1, argv = [], env = {}): # Load kernel self.slide = 0 - self.loading_file = self.kernel - kern_cmds = self.kernel.commands self.kernel_base = None - for cmd in kern_cmds: - if cmd.cmd_id == LC_SEGMENT_64: - if self.kernel_base is None: - self.kernel_base = cmd.vm_address - self.loadSegment64(cmd, False) + for seg in self.kernel.segments: + if self.kernel_base is None: + self.kernel_base = seg.virtual_address + self.loadSegment64(seg, False) self.ql.log.info("Kernel loaded at 0x%x" % self.kernel_base) # Resolve local relocation - for relocation in self.macho_file.dysymbol_table.locreloc: + with open(self.ql.path, 'rb') as _f: + self._macho_raw = _f.read() + dsc = self.macho_file.dynamic_symbol_command + all_secs = list(self.macho_file.sections) + locrelocs = _parse_macho_dysymtab_relocs( + self._macho_raw, + dsc.local_relocation_offset, + dsc.nb_local_relocations + ) + for relocation in locrelocs: + # symbolnum is a 1-indexed section number for local relocations + sec_idx = relocation['symbolnum'] - 1 seg = None - for segment in self.macho_file.segments: - if relocation.symbolnum in segment.sections_index: - seg = segment - break - current_value, = struct.unpack("= len(all_syms): + continue + symname = all_syms[sym_idx].name.encode() + if relocation['length'] == 2 and relocation['rtype'] == 2: if symname not in self.static_symbols: if symname in self.kernel_local_symbols_detail: real_addr = self.kernel_local_symbols_detail[symname]["n_value"] @@ -250,44 +305,48 @@ def loadDriver(self, stack_addr, loadbase = -1, argv = [], env = {}): self.ql.mem.write(self.static_addr + offset, jmpcode) - self.ql.mem.write(loadbase + relocation.address, struct.pack(" 0: + n_sect = sym.numberof_sections # 1-indexed + sec = all_secs_kext[n_sect - 1] if n_sect <= len(all_secs_kext) else None + if sec and sec.name.rstrip('\x00') == '__const': + symname = sym.name.encode() + self.ql.log.info("Found vtable of %s at 0x%x" % (symname, loadbase + sym.value)) + self.vtables[symname] = loadbase + sym.value kext = self.plist["IOKitPersonalities"][self.kext_name]["IOClass"] user = self.plist["IOKitPersonalities"][self.kext_name]["IOUserClientClass"] @@ -305,12 +364,13 @@ def loadDriver(self, stack_addr, loadbase = -1, argv = [], env = {}): self.user_attach = None self.user_start = None - for relocation in self.macho_file.dysymbol_table.extreloc: - symbol = self.macho_file.symbol_table.symbols[relocation.symbolnum] - symname = self.macho_file.string_table[symbol.n_strx] - if b"externalMethod" in symname: - current_value, = struct.unpack(" 5: return - # three pass - # 1: unixthread, uuid, code signature - # 2: segment - # 3: dyld - for pass_count in range(1, 4): + binary = self.dyld_file if isdyld else self.macho_file - if isdyld: - cmds = self.dyld_file.commands - else: - cmds = self.macho_file.commands - - for cmd in cmds: - if pass_count == 1: - if cmd.cmd_id == LC_UNIXTHREAD: - self.loadUnixThread(cmd, isdyld) - - if cmd.cmd_id == LC_UUID: - self.loadUuid() - - if cmd.cmd_id == LC_CODE_SIGNATURE: - self.loadCodeSignature() - - if cmd.cmd_id == LC_MAIN: - self.loadMain(cmd) - - if pass_count == 2: - if cmd.cmd_id == LC_SEGMENT: - pass - - if cmd.cmd_id == LC_SEGMENT_64: - self.loadSegment64(cmd, isdyld) - - if pass_count == 3: - if cmd.cmd_id == LC_LOAD_DYLINKER: - self.loadDylinker(cmd) - self.using_dyld = True - if not isdyld: - if not self.dyld_path: - raise QlErrorMACHOFormat("Error No Dyld path") - self.dyld_path = os.path.join(self.ql.rootfs + self.dyld_path) - self.dyld_file = MachoParser(self.ql, self.dyld_path) - self.loading_file = self.dyld_file - self.proc_entry = self.loadMacho(depth + 1, True) - self.loading_file = self.macho_file - self.using_dyld = True + # Pass 1: thread/main/uuid/code signature + if binary.thread_command is not None: + self.loadUnixThread(binary.thread_command, isdyld) + elif binary.main_command is not None: + self.loadMain(binary.main_command) + self.loadUuid() + self.loadCodeSignature() + + # Pass 2: segments + for seg in binary.segments: + self.loadSegment64(seg, isdyld) + + # Pass 3: dylinker + if binary.dylinker is not None: + self.loadDylinker(binary.dylinker) + self.using_dyld = True + if not isdyld: + if not self.dyld_path: + raise QlErrorMACHOFormat("Error No Dyld path") + self.dyld_path = os.path.join(self.ql.rootfs + self.dyld_path) + self.dyld_file = _lief_macho_parse(self.dyld_path, self.ql.arch.type) + self.proc_entry = self.loadMacho(depth + 1, True) + self.using_dyld = True if depth == 0: self.mmap_address = mmap_address @@ -424,17 +464,14 @@ def loadMacho(self, depth=0, isdyld=False): return self.proc_entry - def loadSegment64(self, cmd, isdyld): + def loadSegment64(self, seg, isdyld): PAGE_SIZE = 0x1000 - if isdyld: - slide = self.dyld_slide - else: - slide = self.slide - vaddr_start = cmd.vm_address + slide - vaddr_end = cmd.vm_address + cmd.vm_size + slide - seg_size = cmd.vm_size - seg_name = cmd.segment_name - seg_data = bytes(self.loading_file.get_segment(seg_name).content) + slide = self.dyld_slide if isdyld else self.slide + vaddr_start = seg.virtual_address + slide + vaddr_end = seg.virtual_address + seg.virtual_size + slide + seg_size = seg.virtual_size + seg_name = seg.name + seg_data = bytes(seg.content) if seg_size == 0: return -1 @@ -460,11 +497,11 @@ def loadSegment64(self, cmd, isdyld): return vaddr_start def loadUnixThread(self, cmd, isdyld): + entry = cmd.pc if not isdyld: - self.binary_entry = cmd.entry - - self.proc_entry = cmd.entry - self.ql.log.debug("Binary Thread Entry: {}".format(hex(cmd.entry))) + self.binary_entry = entry + self.proc_entry = entry + self.ql.log.debug("Binary Thread Entry: {}".format(hex(entry))) def loadUuid(self): @@ -476,10 +513,10 @@ def loadCodeSignature(self): pass def loadMain(self, cmd, isdyld=False): - if self.macho_file.page_zero_size: + if self._page_zero_size: if not isdyld: - self.binary_entry = cmd.entry_offset + self.macho_file.page_zero_size - self.proc_entry = cmd.entry_offset + self.macho_file.page_zero_size + self.binary_entry = cmd.entrypoint + self._page_zero_size + self.proc_entry = cmd.entrypoint + self._page_zero_size def loadDylinker(self, cmd): self.dyld_path = cmd.name @@ -553,9 +590,7 @@ def loadStack(self): if self.using_dyld: ptr -= 4 - #ql.log.info("Binary Dynamic Entry Point: {:X}".format(self.binary_entry)) - self.push_stack_addr(self.macho_file.header_address) - # self.push_stack_addr(self.binary_entry) + self.push_stack_addr(self._page_zero_size) # header_address == page_zero_size return self.stack_sp diff --git a/qiling/loader/macho_parser/__init__.py b/qiling/loader/macho_parser/__init__.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/qiling/loader/macho_parser/const.py b/qiling/loader/macho_parser/const.py deleted file mode 100644 index 0726f374d..000000000 --- a/qiling/loader/macho_parser/const.py +++ /dev/null @@ -1,59 +0,0 @@ - -#!/usr/bin/env python3 -# -# Cross Platform and Multi Architecture Advanced Binary Emulation Framework -# - -# https://opensource.apple.com/source/cctools/cctools-795/include/mach-o/loader.h - -# magic -# MAGIC_32 = 0xFEEDFACE -# MAGIC_64 = 0xFEEDFACF -# MAGIC_FAT = 0xBEBAFECA -MAGIC_32 = [0xFEEDFACE, 0xCEFAEDFE] -MAGIC_64 = [0xFEEDFACF, 0xCFFAEDFE] -MAGIC_FAT = [0xBEBAFECA, 0xCAFEBABE] - -# cpu type -CPU_TYPE_X8664 = 0x01000007 -CPU_TYPE_ARM64 = 0x0100000C -CPU_TYPE_X86 = 0x00000007 -CPU_SUBTYPE_ARM64_ALL = 0x00000006 -CPU_SUBTYPE_X8664_ALL = 0x00000030 -CPU_SUBTYPE_i386_ALL = 0x00000030 - -# file type -MH_DYLINKER = 0x00000007 -MH_EXECUTE = 0x00000002 - -# load command -LC_SEGMENT_64 = 0x00000019 -LC_SEGMENT = 0x00000001 -LC_SYMTAB = 0x00000002 -LC_DYSYMTAB = 0x0000000B -LC_ID_DYLINKER = 0x0000000F -LC_UUID = 0x0000001B -LC_VERSION_MIN_MACOSX = 0x00000024 -LC_VERSION_MIN_IPHONEOS = 0x00000025 -LC_SOURCE_VERSION = 0x0000002A -LC_UNIXTHREAD = 0x00000005 -LC_SEGMENT_SPLIT_INFO = 0x0000001E -LC_FUNCTION_STARTS = 0x00000026 -LC_DATA_IN_CODE = 0x00000029 -LC_CODE_SIGNATURE = 0x0000001D -LC_DYLD_INFO_ONLY = 0x80000022 -LC_LOAD_DYLINKER = 0x0000000E -LC_MAIN = 0x80000028 -LC_LOAD_DYLIB = 0x0000000C -LC_LOAD_WEAK_DYLIB = 0x80000018 -LC_ENCRYPTION_INFO_64 = 0x0000002C -LC_BUILD_VERSION = 0x00000032 -LC_DYLD_EXPORTS_TRIE = 0x80000033 -LC_DYLD_CHAINED_FIXUPS = 0x80000034 -LC_RPATH = 0x8000001C -LC_ID_DYLIB = 0x0000000D - -# UNIXTHREAD -X86_THREAD_STATE32 = 0x00000001 -X86_THREAD_STATE64 = 0x00000004 -ARM_THREAD_STATE64 = 0x00000006 \ No newline at end of file diff --git a/qiling/loader/macho_parser/data.py b/qiling/loader/macho_parser/data.py deleted file mode 100644 index 7acda503a..000000000 --- a/qiling/loader/macho_parser/data.py +++ /dev/null @@ -1,237 +0,0 @@ -#!/usr/bin/env python3 -# -# Cross Platform and Multi Architecture Advanced Binary Emulation Framework -# - -from struct import unpack - -class Segment: - - def __init__(self, lc, data): - self.name = lc.segment_name - self.section_num = lc.number_of_sections - self.vm_address = lc.vm_address - self.vm_size = lc.vm_size - self.file_offset = lc.file_offset - self.file_size = lc.file_size - self.max_vm_protection = lc.maximum_vm_protection - self.init_vm_protection = lc.initial_vm_protection - self.flags = lc.flags - self.content = data[self.file_offset : self.file_offset + self.file_size] - self.sections = [] - self.sections_index = [] - for i in range(self.section_num): - self.sections.append(Section(lc.sections[i], data)) - - # def __str__(self): - # return (" Segment {}: content {}".format(self.name, self.content)) - - -class Section: - - def __init__(self, lc, data): - self.name = lc.section_name - self.segment_name = lc.segment_name - self.address = lc.address - self.size = lc.size - self.offset = lc.offset - self.align = lc.alignment - self.rel_offset = lc.relocations_offset - self.rel_num = lc.number_of_relocations - self.flags = lc.flags - self.content = data[self.offset : self.offset + self.size] - - # def __str__(self): - # return (" Section {}: content {}".format(self.name,self.content)) - - -class FunctionStarts: - - def __init__(self, lc, data): - self.offset = lc.data_offset - self.size = lc.data_size - self.content = data[self.offset : self.offset + self.size] - - # def __str__(self): - # return (" FunctionStarts: content {}".format(self.content)) - - -class Symbol64(object): - def __init__(self, data): - self.n_strx = unpack(">= 24 - self.pcrel = tmp & 0x1 - tmp >>= 1 - self.length = tmp & 0x3 - tmp >>= 2 - self.extern = tmp & 0x1 - tmp >>= 1 - self.rtype = tmp & 0xf - - def __str__(self): - return """Relocation header: - address = {} - symbolnum = {} - pcrel = {} - length = {} - extern = {} - rtype = {} - """.format(hex(self.address), self.symbolnum, self.pcrel, self.length, self.extern, self.rtype) - - -class DySymbolTable: - # TODO: finish parser - def __init__(self, lc, data): - self.locsymbol_index = lc.locsymbol_index - self.locsymbol_num = lc.locsymbol_number - self.defext_index = lc.defined_extsymbol_index - self.defext_num = lc.defined_extsymbol_number - self.undefext_index = lc.undef_extsymbol_index - self.undefext_num = lc.undef_extsymbol_number - self.indsym_offset = lc.indsym_table_offset - self.indsym_num = lc.indsym_table_entries - self.indirect_symbols = [] - self.extreloc_offset = lc.extreloc_table_offset - self.extreloc_num = lc.extreloc_table_entries - self.extreloc = [] - self.locreloc_offset = lc.locreloc_table_offset - self.locreloc_num = lc.locreloc_table_entries - self.locreloc = [] - - if self.indsym_num: - slide = 0 - for i in range(self.indsym_num): - self.indirect_symbols.append(unpack("L", FR.read(4))[0] - for i in range(self.arch_num): - FI = FatInfo(FR.read(4 * 5)) - self.binarys.append(FI) - - def getBinary(self, arch): - - for item in self.binarys: - if item.cpu_type == CPU_TYPE_X8664: - return item - elif item.cpu_type == CPU_TYPE_ARM64: - return item - return None - -class FatInfo: - def __init__(self, data): - FR = FileReader(data) - self.cpu_type = unpack(">L", FR.read(4))[0] - self.cpu_subtype = unpack(">L", FR.read(4))[0] - self.offset = unpack(">L", FR.read(4))[0] - self.size = unpack(">L", FR.read(4))[0] - self.align = 2 ** unpack(">L", FR.read(4))[0] - - # def __str__(self): - # return ("CPU 0x%X, CPU subtype 0x%X, offset 0x%X, size 0x%X, align %d" %( - # self.cpu_type, self.cpu_subtype, self.offset, self.size, self.align - # )) \ No newline at end of file diff --git a/qiling/loader/macho_parser/loadcommand.py b/qiling/loader/macho_parser/loadcommand.py deleted file mode 100644 index 1eb18d362..000000000 --- a/qiling/loader/macho_parser/loadcommand.py +++ /dev/null @@ -1,540 +0,0 @@ -#!/usr/bin/env python3 -# -# Cross Platform and Multi Architecture Advanced Binary Emulation Framework -# - -from struct import unpack - -from .const import * -from .utils import * - -# TODO: We need support more LC command to load more kinds of binary - -class LoadCommand: - - def __init__(self, data): - self.data = data - self.FR = FileReader(data) - self.cmd_id = unpack("= 8: - lc = LoadCommand(self.lc_raw[offset:]) - else: - self.ql.log.info("cmd size overflow") - return False - - if self.header.lc_size >= offset + lc.cmd_size: - complete_cmd = lc.get_complete() - pass - else: - self.ql.log.info("cmd size overflow") - return False - - self.commands.append(complete_cmd) - - offset += lc.cmd_size - - return True - - - def parseData(self): - self.segments = [] - self.sections = [None] - for command in self.commands: - if command.cmd_id == LC_SEGMENT_64: - tmp = Segment(command, self.binary_file) - tmp.sections_index += range(len(self.sections), len(self.sections) + len(tmp.sections)) - self.segments.append(tmp) - for section in tmp.sections: - self.sections.append(section) - elif command.cmd_id == LC_SEGMENT: - tmp = Segment(command, self.binary_file) - self.segments.append(tmp) - for section in tmp.sections: - self.sections.append(section) - elif command.cmd_id == LC_FUNCTION_STARTS: - self.function_starts = FunctionStarts(command, self.binary_file) - elif command.cmd_id == LC_SYMTAB: - self.symbol_table = SymbolTable(command, self.binary_file) - self.string_table = StringTable(command, self.binary_file) - elif command.cmd_id == LC_DATA_IN_CODE: - self.data_in_code = DataInCode(command, self.binary_file) - elif command.cmd_id == LC_CODE_SIGNATURE: - self.code_signature = CodeSignature(command, self.binary_file) - elif command.cmd_id == LC_SEGMENT_SPLIT_INFO: - self.seg_split_info = SegmentSplitInfo(command, self.binary_file) - elif command.cmd_id == LC_DYSYMTAB: - self.dysymbol_table = DySymbolTable(command, self.binary_file) - return True - - @staticmethod - def getMagic(binary): - return unpack(" bytearray: + """Build a flat virtual-address-mapped PE image.""" + oh = binary.optional_header + data = bytearray(oh.sizeof_image) + data[:oh.sizeof_headers] = raw_bytes[:oh.sizeof_headers] + for sec in binary.sections: + va = sec.virtual_address + content = bytes(sec.content) + end = min(va + len(content), oh.sizeof_image) + if end > va: + data[va:end] = content[:end - va] + return data + + +def _pe_apply_relocations(data: bytearray, binary: PE.Binary, new_base: int) -> None: + """Apply base-relocation delta in-place.""" + delta = new_base - binary.optional_header.imagebase + if delta == 0: + return + BT = PE.RelocationEntry.BASE_TYPES + for block in binary.relocations: + for entry in block.entries: + rva = block.virtual_address + entry.position + if entry.type == BT.HIGHLOW: + val = int.from_bytes(data[rva:rva + 4], 'little') + data[rva:rva + 4] = ((val + delta) & 0xFFFFFFFF).to_bytes(4, 'little') + elif entry.type == BT.DIR64: + val = int.from_bytes(data[rva:rva + 8], 'little') + data[rva:rva + 8] = ((val + delta) & 0xFFFFFFFFFFFFFFFF).to_bytes(8, 'little') + + +def _pe_map_image(ql: Qiling, binary: PE.Binary, image_base: int, image_size: int, image_name: str, data: bytearray, *, skip_discardable: bool = False) -> None: + """Map a PE image, using per-section permissions when possible.""" + + sec_alignment = binary.optional_header.section_alignment + headers_size = min(len(data), binary.optional_header.sizeof_headers) + + def __map_sections() -> None: + hdr_size = ql.mem.align_up(headers_size, sec_alignment) + + if hdr_size: + ql.mem.map(image_base, hdr_size, info=image_name) + ql.mem.write(image_base, bytes(data[:headers_size])) + + SC = PE.Section.CHARACTERISTICS + + for section in binary.sections: + chars = int(section.characteristics) + + if skip_discardable and (chars & int(SC.MEM_DISCARDABLE)): + continue + + sec_base = image_base + section.virtual_address + sec_vsize = max(section.virtual_size, len(bytes(section.content))) + sec_size = ql.mem.align_up(sec_vsize, sec_alignment) + + if sec_size == 0: + continue + + sec_perm = 0 + + if chars & int(SC.MEM_READ): + sec_perm |= 1 + if chars & int(SC.MEM_WRITE): + sec_perm |= 2 + if chars & int(SC.MEM_EXECUTE): + sec_perm |= 4 + if sec_perm == 0: + sec_perm = 1 + + sec_name = section.name.rstrip('\x00') or 'section' + ql.mem.map(sec_base, sec_size, sec_perm, f'{image_name} ({sec_name})') + + start = section.virtual_address + stop = start + min(sec_vsize, len(data) - start) + + if stop > start: + ql.mem.write(sec_base, bytes(data[start:stop])) + + def __map_all() -> None: + ql.mem.map(image_base, image_size, info=image_name) + ql.mem.write(image_base, bytes(data)) + + if sec_alignment and (sec_alignment % ql.mem.pagesize) == 0: + __map_sections() + else: + __map_all() class QlPeCacheEntry(NamedTuple): @@ -92,10 +180,12 @@ class Process: # maps image base to its list of function table entries function_tables: MutableMapping[int, List] + function_table_entry_size: Dict[int, int] # List of exports which have been forwarded from # one DLL to another. forwarded_exports: List[ForwardedExport] + resolved_forwarded_exports: Set[ForwardedExport] def __init__(self, ql: Qiling): self.ql = ql @@ -124,35 +214,33 @@ def __get_path_elements(self, name: str) -> Tuple[str, str]: return self.ql.os.path.virtual_to_host_path(vpath), basename.casefold() - def init_function_tables(self, pe: pefile.PE, image_base: int): + def init_function_tables(self, pe: PE.Binary, image_base: int): """Parse function table data for the given PE file. - Only really relevant for non-x86 images. + Only relevant for non-x86 images. Args: pe: the PE image whose function data should be parsed image_base: the absolute address at which the image was loaded """ - if self.ql.arch.type is not QL_ARCH.X86: + if self.ql.arch.type is QL_ARCH.X86: + return - # Check if the PE file has an exception directory - if hasattr(pe, 'DIRECTORY_ENTRY_EXCEPTION'): - exception_dir = pe.OPTIONAL_HEADER.DATA_DIRECTORY[ - pefile.DIRECTORY_ENTRY['IMAGE_DIRECTORY_ENTRY_EXCEPTION'] - ] - - self.function_table_lookup[image_base] = exception_dir.VirtualAddress + exc_dir = pe.data_directory(PE.DataDirectory.TYPES.EXCEPTION_TABLE) - runtime_function_list = list(pe.DIRECTORY_ENTRY_EXCEPTION) + if exc_dir is None or exc_dir.rva == 0: + self.ql.log.debug('Image has no exception directory; skipping exception data') + return - if image_base not in self.function_tables: - self.function_tables[image_base] = [] + self.function_table_lookup[image_base] = exc_dir.rva + self.function_table_entry_size[image_base] = 12 if self.ql.arch.type is QL_ARCH.X8664 else 8 - self.function_tables[image_base].extend(runtime_function_list) + if image_base not in self.function_tables: + self.function_tables[image_base] = [] - self.ql.log.debug(f'Parsed {len(runtime_function_list)} exception directory entries') + runtime_functions = list(pe.exception_functions) if hasattr(pe, 'exception_functions') else [] + self.function_tables[image_base].extend(runtime_functions) - else: - self.ql.log.debug(f'Image has no exception directory; skipping exception data') + self.ql.log.debug(f'Parsed {len(runtime_functions)} exception directory entries') def lookup_function_entry(self, base_addr: int, control_pc: int): """Look up a RUNTIME_FUNCTION entry and its index in a module's @@ -170,14 +258,21 @@ def lookup_function_entry(self, base_addr: int, control_pc: int): # Initiate a search of the function table for a RUNTIME_FUNCTION # entry such that the provided PC falls within its start and end range. - return next(((i, rtfunc) for i, rtfunc in enumerate(function_table) - if rtfunc.struct.BeginAddress <= control_pc - base_addr < rtfunc.struct.EndAddress), - (None, None)) + return next( + ((i, rtfunc) for i, rtfunc in enumerate(function_table) + if rtfunc.rva_start <= control_pc - base_addr < rtfunc.rva_end), + (None, None) + ) def resolve_forwarded_exports(self): + pending_forwarded_exports = [] + while self.forwarded_exports: forwarded_export = self.forwarded_exports.pop() + if forwarded_export in self.resolved_forwarded_exports: + continue + source_dll = forwarded_export.source_dll source_ordinal = forwarded_export.source_ordinal source_symbol = forwarded_export.source_symbol @@ -194,7 +289,7 @@ def resolve_forwarded_exports(self): target_iat = self.import_address_table.get(target_dll) if not target_iat: - # If IAT was not found, it is probably a virtual library. + pending_forwarded_exports.append(forwarded_export) continue # If we have an existing entry in the process IAT for the code @@ -203,7 +298,7 @@ def resolve_forwarded_exports(self): forward_ea = target_iat.get(target_symbol) if not forward_ea: - self.ql.log.warning(f"Forwarding symbol {source_dll}.{source_symbol} to {target_dll}.{target_symbol}: Failed to resolve address") + pending_forwarded_exports.append(forwarded_export) continue self.import_address_table[source_dll][source_symbol] = forward_ea @@ -224,8 +319,30 @@ def resolve_forwarded_exports(self): # The most correct way to resolve this would be to add # support for addresses to be associated with multiple symbols. + self.resolved_forwarded_exports.add(forwarded_export) self.ql.log.debug(f"Forwarding symbol {source_dll}.{source_symbol} to {target_dll}.{target_symbol}: Resolved symbol to ({forward_ea:#x})") + self.forwarded_exports = pending_forwarded_exports + + def queue_forwarded_exports(self, dll: PE.Binary, dll_name: str) -> None: + if not dll.has_exports: + return + + dll_export = dll.get_export() + + for sym in (dll_export.entries if dll_export else []): + if not sym.is_forwarded: + continue + + fi = sym.forward_information + self.forwarded_exports.append(ForwardedExport( + source_dll=dll_name, + source_ordinal=sym.ordinal, + source_symbol=sym.name.encode() if sym.name else None, + target_dll=(fi.library.lower() + '.dll').casefold(), + target_symbol=fi.function.encode() + )) + def load_dll(self, name: str, is_driver: bool = False) -> int: dll_path, dll_name = self.__get_path_elements(name) @@ -260,6 +377,7 @@ def load_dll(self, name: str, is_driver: bool = False) -> int: cached = None loaded = False + dll = None if self.libcache: cached = self.libcache.restore(dll_path) @@ -280,23 +398,29 @@ def load_dll(self, name: str, is_driver: bool = False) -> int: for entry in cached.cmdlines: self.set_cmdline(entry['name'], entry['address'], data) - self.ql.log.info(f'Loaded {dll_name} from cache') - loaded = True + dll = PE.parse(dll_path) + + if dll is None: + self.ql.log.warning(f'Failed to re-parse cached DLL metadata: {dll_path}') + else: + self.init_function_tables(dll, image_base) + self.queue_forwarded_exports(dll, dll_name) + self.ql.log.info(f'Loaded {dll_name} from cache') + loaded = True # either file was not cached, or could not be loaded to the same location in memory if not cached or not loaded: - dll = pefile.PE(dll_path, fast_load=True) - dll.parse_data_directories() - warnings = dll.get_warnings() + with open(dll_path, 'rb') as f: + dll_raw = f.read() - if warnings: - self.ql.log.debug(f'Warnings while loading {dll_name}:') + dll = PE.parse(dll_path) - for warning in warnings: - self.ql.log.debug(f' - {warning}') + if dll is None: + self.ql.log.error(f'Failed to parse PE: {dll_path}') + return 0 - image_base = dll.OPTIONAL_HEADER.ImageBase or self.dll_last_address - image_size = self.ql.mem.align_up(dll.OPTIONAL_HEADER.SizeOfImage) + image_base = dll.optional_header.imagebase or self.dll_last_address + image_size = self.ql.mem.align_up(dll.optional_header.sizeof_image) relocate = False self.ql.log.debug(f'DLL preferred base address: {image_base:#x}') @@ -311,60 +435,41 @@ def load_dll(self, name: str, is_driver: bool = False) -> int: self.ql.log.debug(f'DLL preferred base address is taken, loading to: {image_base:#x}') relocate = True + data = _pe_build_mapped_image(dll, dll_raw) + if relocate: with ShowProgress(self.ql.log, 0.1337): - dll.relocate_image(image_base) + _pe_apply_relocations(data, dll, image_base) # initialize the function tables only after possible relocation self.init_function_tables(dll, image_base) - data = bytearray(dll.get_memory_mapped_image()) assert image_size >= len(data) cmdlines = [] - for sym in dll.DIRECTORY_ENTRY_EXPORT.symbols: - ea = image_base + sym.address + if dll.has_exports: + self.queue_forwarded_exports(dll, dll_name) + dll_export = dll.get_export() - if sym.forwarder: - # Some exports are forwarders, meaning they - # actually refer to code in other libraries. - # - # For example, calls to - # kernel32.InterlockedPushEntrySList - # should be forwarded to - # ntdll.RtlInterlockedPushEntrySList - # - # If we do not properly account for forwarders then - # calls to these symbols will land in the exporter's - # data section and cause a lot of problems. - forward_str = sym.forwarder + for sym in (dll_export.entries if dll_export else []): + sym_name = sym.name.encode() if sym.name else None + ea = image_base + sym.address - if b'.' in forward_str: - target_dll_name, target_symbol_name = forward_str.split(b'.', 1) + import_symbols[ea] = { + 'name' : sym_name, + 'ordinal' : sym.ordinal, + 'dll' : dll_name.split('.')[0] + } - target_dll_filename = (target_dll_name.lower() + b'.dll').decode() + if sym_name: + import_table[sym_name] = ea - # Remember the forwarded export for later. - forwarded_export = ForwardedExport(dll_name, sym.ordinal, sym.name, - target_dll_filename, target_symbol_name) + import_table[sym.ordinal] = ea + cmdline_entry = self.set_cmdline(sym_name, sym.address, data) - self.forwarded_exports.append(forwarded_export) - - import_symbols[ea] = { - 'name' : sym.name, - 'ordinal' : sym.ordinal, - 'dll' : dll_name.split('.')[0] - } - - if sym.name: - import_table[sym.name] = ea - - import_table[sym.ordinal] = ea - cmdline_entry = self.set_cmdline(sym.name, sym.address, data) - - if cmdline_entry: - cmdlines.append(cmdline_entry) + if cmdline_entry: + cmdlines.append(cmdline_entry) if self.libcache: cached = QlPeCacheEntry(image_base, data, cmdlines, import_symbols, import_table) @@ -381,8 +486,8 @@ def load_dll(self, name: str, is_driver: bool = False) -> int: dll_len = image_size self.dll_size += dll_len - self.ql.mem.map(dll_base, dll_len, info=dll_name) - self.ql.mem.write(dll_base, bytes(data)) + assert dll is not None + _pe_map_image(self.ql, dll, dll_base, dll_len, dll_name, data) if dll_base == self.dll_last_address: self.dll_last_address = self.ql.mem.align_up(self.dll_last_address + dll_len, 0x10000) @@ -397,7 +502,7 @@ def load_dll(self, name: str, is_driver: bool = False) -> int: if not cached or not loaded: # parse directory entry import self.ql.log.debug(f'Init imports for {dll_name}') - self.init_imports(dll, is_driver) + self.init_imports(dll, is_driver, image_base) # calling DllMain is essential for dlls to initialize properly. however # DllMain of system libraries may fail due to incomplete or inaccurate @@ -414,10 +519,10 @@ def load_dll(self, name: str, is_driver: bool = False) -> int: return dll_base - def call_dll_entrypoint(self, dll: pefile.PE, dll_base: int, dll_len: int, dll_name: str): - entry_address = dll.OPTIONAL_HEADER.AddressOfEntryPoint + def call_dll_entrypoint(self, dll: PE.Binary, dll_base: int, dll_len: int, dll_name: str): + entry_address = dll.optional_header.addressof_entrypoint - if dll.get_section_by_rva(entry_address) is None: + if dll.section_from_rva(entry_address) is None: return if dll_name in ('kernelbase.dll', 'kernel32.dll'): @@ -608,19 +713,21 @@ def populate_unistr(obj, s: str) -> None: self.ldr_list.append(entry_addr) @staticmethod - def directory_exists(pe: pefile.PE, entry: str) -> bool: - ent = pefile.DIRECTORY_ENTRY[entry] - - return pe.OPTIONAL_HEADER.DATA_DIRECTORY[ent].VirtualAddress != 0 - - def init_imports(self, pe: pefile.PE, is_driver: bool): + def directory_exists(pe: PE.Binary, entry: str) -> bool: + if entry == 'IMAGE_DIRECTORY_ENTRY_IMPORT': + return pe.has_imports + elif entry == 'IMAGE_DIRECTORY_ENTRY_EXPORT': + return pe.has_exports + elif entry == 'IMAGE_DIRECTORY_ENTRY_LOAD_CONFIG': + return pe.load_configuration is not None + return False + + def init_imports(self, pe: PE.Binary, is_driver: bool, image_base: int = 0): if not Process.directory_exists(pe, 'IMAGE_DIRECTORY_ENTRY_IMPORT'): return - pe.full_load() - - for entry in pe.DIRECTORY_ENTRY_IMPORT: - dll_name = entry.dll.decode().casefold() + for entry in pe.imports: + dll_name = entry.name.casefold() self.ql.log.debug(f'Requesting imports from {dll_name}') orig_dll_name = dll_name @@ -647,17 +754,21 @@ def init_imports(self, pe: pefile.PE, is_driver: bool): 'kernel32.dll', 'ntdll.dll', 'kernelbase.dll', - 'ucrtbase.dll' + 'ucrtbase.dll', + 'user32.dll', + 'gdi32.dll', + 'win32u.dll' ) - imports = iter(entry.imports) + imports = iter(entry.entries) failed = False fallback = None while not redirected and not failed: # find all possible redirection options by scanning key dlls for the current imported symbol imp = next(imports, None) - redirection_options = [fallback] if imp is None else [filename for filename in key_dlls if filename in self.import_address_table and imp.name in self.import_address_table[filename]] + imp_name_bytes = (imp.name.encode() if isinstance(imp.name, str) else imp.name) if imp and imp.name else None + redirection_options = [fallback] if imp is None else [filename for filename in key_dlls if filename in self.import_address_table and imp_name_bytes in self.import_address_table[filename]] # no redirection options: failed to redirect dll if not redirection_options: @@ -679,54 +790,51 @@ def init_imports(self, pe: pefile.PE, is_driver: bool): self.ql.log.debug(f'Redirecting {dll_name} to {key_dll}') dll_name = key_dll - unbound_imports = [imp for imp in entry.imports if not imp.bound] + all_imports = list(entry.entries) + lookup_dll_name = dll_name - if unbound_imports: - # Only load dll if encountered unbound symbol + if all_imports: + # Only load dll if there are imports to resolve if not redirected: - dll_base = self.load_dll(entry.dll.decode(), is_driver) + dll_base = self.load_dll(lookup_dll_name, is_driver) if not dll_base: continue - for imp in unbound_imports: - iat = self.import_address_table[dll_name] + for imp in all_imports: + iat = self.import_address_table[lookup_dll_name] + imp_name_bytes = (imp.name.encode() if isinstance(imp.name, str) else imp.name) if imp.name else None - if imp.name: - if imp.name not in iat: - self.ql.log.debug(f'Error in loading function {imp.name.decode()} ({orig_dll_name}){", probably misdirected" if redirected else ""}') + if imp_name_bytes: + if imp_name_bytes not in iat: + self.ql.log.debug(f'Error in loading function {imp.name} ({orig_dll_name}){", probably misdirected" if redirected else ""}') continue - addr = iat[imp.name] + addr = iat[imp_name_bytes] else: addr = iat[imp.ordinal] - self.ql.mem.write_ptr(imp.address, addr) + self.ql.mem.write_ptr(image_base + imp.iat_address, addr) - def init_exports(self, pe: pefile.PE): + def init_exports(self, pe: PE.Binary): if not Process.directory_exists(pe, 'IMAGE_DIRECTORY_ENTRY_EXPORT'): return - # Do a full load if IMAGE_DIRECTORY_ENTRY_EXPORT is present so we can load the exports - pe.full_load() - - # address corner case for malformed export tables where IMAGE_DIRECTORY_ENTRY_EXPORT exists, but DIRECTORY_ENTRY_EXPORT does not - if not hasattr(pe, 'DIRECTORY_ENTRY_EXPORT'): - return iat = {} + pe_export = pe.get_export() - # parse directory entry export - for entry in pe.DIRECTORY_ENTRY_EXPORT.symbols: + for entry in (pe_export.entries if pe_export else []): ea = self.pe_image_address + entry.address + entry_name = entry.name.encode() if entry.name else None self.export_symbols[ea] = { - 'name' : entry.name, + 'name' : entry_name, 'ordinal' : entry.ordinal } - if entry.name: - iat[entry.name] = ea + if entry_name: + iat[entry_name] = ea iat[entry.ordinal] = ea @@ -800,11 +908,11 @@ def init_ki_user_shared_data(self): self.ql.os.KUSER_SHARED_DATA = kusd_obj - def init_security_cookie(self, pe: pefile.PE, image_base: int): + def init_security_cookie(self, pe: PE.Binary, image_base: int): if not Process.directory_exists(pe, 'IMAGE_DIRECTORY_ENTRY_LOAD_CONFIG'): return - cookie_rva = pe.DIRECTORY_ENTRY_LOAD_CONFIG.struct.SecurityCookie - pe.OPTIONAL_HEADER.ImageBase + cookie_rva = pe.load_configuration.security_cookie - pe.optional_header.imagebase # get a random cookie value but keep the two most significant bytes zeroes # @@ -840,10 +948,17 @@ def run(self): if self.ql.code: pe = None + pe_raw = None self.is_driver = False else: - pe = pefile.PE(self.path, fast_load=True) - self.is_driver = pe.is_driver() + with open(self.path, 'rb') as f: + pe_raw = f.read() + pe = PE.parse(self.path) + if pe is None: + raise QlErrorArch(f'Failed to parse PE: {self.path}') + C = PE.Header.CHARACTERISTICS + self.is_driver = bool(pe.header.characteristics & int(C.SYSTEM)) or \ + pe.optional_header.subsystem == PE.OptionalHeader.SUBSYSTEM.NATIVE ossection = f'OS{self.ql.arch.bits}' @@ -864,7 +979,9 @@ def run(self): self.ldr_list = [] self.function_tables = {} self.function_table_lookup = {} + self.function_table_entry_size = {} self.forwarded_exports = [] + self.resolved_forwarded_exports = set() self.pe_image_address = 0 self.pe_image_size = 0 self.dll_size = 0 @@ -878,32 +995,35 @@ def run(self): self.cmdline = bytes(f'{cmdline} {cmdargs}\x00', "utf-8") - self.load(pe) + self.load(pe, pe_raw) - def load(self, pe: Optional[pefile.PE]): + def load(self, pe: Optional[PE.Binary], pe_raw: Optional[bytes] = None): # set stack pointer self.ql.log.info("Initiate stack address at 0x%x " % self.stack_address) self.ql.mem.map(self.stack_address, self.stack_size, info="[stack]") if pe is not None: + assert pe_raw is not None image_name = os.path.basename(self.path) - image_base = pe.OPTIONAL_HEADER.ImageBase - image_size = self.ql.mem.align_up(pe.OPTIONAL_HEADER.SizeOfImage) + image_base = pe.optional_header.imagebase + image_size = self.ql.mem.align_up(pe.optional_header.sizeof_image) + + pe_data = _pe_build_mapped_image(pe, pe_raw) # if default base address is taken, use the one specified in profile if not self.ql.mem.is_available(image_base, image_size): image_base = self.image_address - pe.relocate_image(image_base) + _pe_apply_relocations(pe_data, pe, image_base) - self.entry_point = image_base + pe.OPTIONAL_HEADER.AddressOfEntryPoint + self.entry_point = image_base + pe.optional_header.addressof_entrypoint self.pe_image_address = image_base self.pe_image_size = image_size self.ql.log.info(f'Loading {self.path} to {image_base:#x}') self.ql.log.info(f'PE entry point at {self.entry_point:#x}') - self.ql.mem.map(image_base, image_size, info=f'{image_name}') - self.images.append(Image(image_base, image_base + pe.NT_HEADERS.OPTIONAL_HEADER.SizeOfImage, os.path.abspath(self.path))) + _pe_map_image(self.ql, pe, image_base, image_size, image_name, pe_data) + self.images.append(Image(image_base, image_base + pe.optional_header.sizeof_image, os.path.abspath(self.path))) if self.is_driver: self.init_driver_object() @@ -929,15 +1049,13 @@ def load(self, pe: Optional[pefile.PE]): self.init_ki_user_shared_data() - pe.parse_data_directories() - - # done manipulating pe file; write its contents into memory - self.ql.mem.write(image_base, bytes(pe.get_memory_mapped_image())) - if self.is_driver: # security cookie can be written only after image has been loaded to memory self.init_security_cookie(pe, image_base) + # Initialize unwind metadata before any entrypoint code runs. + super().init_function_tables(pe, image_base) + # Stack should not init at the very bottom. Will cause errors with Dlls top_of_stack = self.stack_address + self.stack_size - 0x1000 @@ -961,11 +1079,12 @@ def load(self, pe: Optional[pefile.PE]): # parse directory entry import self.ql.log.debug(f'Init imports for {self.path}') - super().init_imports(pe, self.is_driver) + super().init_imports(pe, self.is_driver, image_base) self.ql.log.debug(f'Done loading {self.path}') - if pe.is_driver(): + C = PE.Header.CHARACTERISTICS + if self.is_driver: args = ( (POINTER, self.driver_object_address), (POINTER, self.regitry_path_address) @@ -975,14 +1094,10 @@ def load(self, pe: Optional[pefile.PE]): self.ql.log.debug(f' PDRIVER_OBJECT DriverObject : {args[0][1]:#010x}') self.ql.log.debug(f' PUNICODE_STRING RegistryPath : {args[1][1]:#010x}') - # We know that a driver will return, so if the user did not configure stop - # options, write a sentinel return value ret = None if self.ql.stop_options else self.ql.stack_write(0, 0xdeadc0de) - - # set up call frame for DriverEntry self.ql.os.fcall.call_native(self.entry_point, args, ret) - elif pe.is_dll(): + elif bool(pe.header.characteristics & int(C.DLL)): args = ( (POINTER, image_base), (DWORD, 1), # DLL_PROCESS_ATTACH @@ -994,12 +1109,7 @@ def load(self, pe: Optional[pefile.PE]): self.ql.log.debug(f' DWORD fdwReason : {args[1][1]:#010x}') self.ql.log.debug(f' LPVOID lpReserved : {args[2][1]:#010x}') - # set up call frame for DllMain self.ql.os.fcall.call_native(self.entry_point, args, None) - - # Initialize the function tables - super().init_function_tables(pe, image_base) - elif pe is None: self.ql.mem.map(self.entry_point, self.ql.os.code_ram_size, info="[shellcode]") @@ -1050,7 +1160,7 @@ class ShowProgress: # animation marker: this is used to tell animation log records from the rest. _marker_ = r'$__ql_anim__' - def __init__(self, logger: Logger, interval: float) -> None: + def __init__(self, logger, interval: float) -> None: from typing import List, Callable from threading import Thread, Event diff --git a/qiling/loader/pe_uefi.py b/qiling/loader/pe_uefi.py index 6ae6f00d9..d3dad7658 100644 --- a/qiling/loader/pe_uefi.py +++ b/qiling/loader/pe_uefi.py @@ -4,15 +4,15 @@ # import os -from typing import Any, Mapping, Optional, Sequence -from pefile import PE +from lief import PE -from unicorn.unicorn_const import UC_PROT_READ, UC_PROT_WRITE, UC_PROT_EXEC +from typing import Any, Mapping, Optional, Sequence from qiling import Qiling from qiling.const import QL_ARCH from qiling.exception import QlErrorArch, QlMemoryMappedError from qiling.loader.loader import QlLoader, Image +from qiling.loader.pe import _pe_apply_relocations, _pe_build_mapped_image, _pe_map_image from qiling.os.const import PARAM_INTN, POINTER from qiling.os.uefi import st, smst, utils @@ -94,72 +94,33 @@ def map_and_load(self, path: str, context: UefiContext, exec_now: bool=False): """ ql = self.ql - pe = PE(path, fast_load=True) - - # use image base only if it does not point to NULL - image_base = pe.OPTIONAL_HEADER.ImageBase or context.next_image_base - image_size = ql.mem.align_up(pe.OPTIONAL_HEADER.SizeOfImage) - image_name = os.path.basename(path) - - assert (image_base % ql.mem.pagesize) == 0, 'image base is expected to be page-aligned' - - if image_base != pe.OPTIONAL_HEADER.ImageBase: - pe.relocate_image(image_base) - # pe.parse_data_directories() + with open(path, 'rb') as f: + pe_raw = f.read() - sec_alignment = pe.OPTIONAL_HEADER.SectionAlignment + pe = PE.parse(path) - def __map_sections(): - """Load file sections to memory, each in its own memory region protected by - its defined permissions. That allows separation of code and data, which makes - it easier to detect abnomal behavior or memory corruptions. - """ + if pe is None: + raise QlMemoryMappedError(f'Failed to parse UEFI module: {path}') - # load the header - hdr_data = bytes(pe.header) - hdr_base = image_base - hdr_size = ql.mem.align_up(len(hdr_data), sec_alignment) - hdr_perm = UC_PROT_READ - - ql.mem.map(hdr_base, hdr_size, hdr_perm, image_name) - ql.mem.write(hdr_base, hdr_data) - - # load sections - for section in pe.sections: - if not section.IMAGE_SCN_MEM_DISCARDABLE: - sec_name = section.Name.rstrip(b'\x00').decode() - sec_data = bytes(section.get_data(ignore_padding=True)) - sec_base = image_base + section.get_VirtualAddress_adj() - sec_size = ql.mem.align_up(len(sec_data), sec_alignment) - - sec_perm = sum(( - section.IMAGE_SCN_MEM_READ * UC_PROT_READ, - section.IMAGE_SCN_MEM_WRITE * UC_PROT_WRITE, - section.IMAGE_SCN_MEM_EXECUTE * UC_PROT_EXEC - )) - - ql.mem.map(sec_base, sec_size, sec_perm, f'{image_name} ({sec_name})') - ql.mem.write(sec_base, sec_data) + # use image base only if it does not point to NULL + image_base = pe.optional_header.imagebase or context.next_image_base + image_size = ql.mem.align_up(pe.optional_header.sizeof_image) + image_name = os.path.basename(path) - def __map_all(): - """Load the entire file to memory as a single memory region. - """ + assert (image_base % ql.mem.pagesize) == 0, 'image base is expected to be page-aligned' - data = bytes(pe.get_memory_mapped_image()) + pe_data = _pe_build_mapped_image(pe, pe_raw) - ql.mem.map(image_base, image_size, info=image_name) - ql.mem.write(image_base, data) + # apply relocations once to the flat image, before either mapping strategy uses it + if image_base != pe.optional_header.imagebase: + _pe_apply_relocations(pe_data, pe, image_base) - # if sections are aligned to page, we can map them separately - if (sec_alignment % ql.mem.pagesize) == 0: - __map_sections() - else: - __map_all() + _pe_map_image(ql, pe, image_base, image_size, image_name, pe_data, skip_discardable=True) ql.log.info(f'Module {path} loaded to {image_base:#x}') - entry_point = image_base + pe.OPTIONAL_HEADER.AddressOfEntryPoint + entry_point = image_base + pe.optional_header.addressof_entrypoint ql.log.info(f'Module entry point at {entry_point:#x}') # the 'entry_point' member is used by the debugger. if not set, set it diff --git a/qiling/os/windows/dlls/ntdll.py b/qiling/os/windows/dlls/ntdll.py index 16956f73f..a6a972a30 100644 --- a/qiling/os/windows/dlls/ntdll.py +++ b/qiling/os/windows/dlls/ntdll.py @@ -19,6 +19,10 @@ from unicorn.x86_const import * + +def _runtime_function_entry_size(ql: Qiling, image_base: int) -> int: + return ql.loader.function_table_entry_size.get(image_base, 12) + # void *memcpy( # void *dest, # const void *src, @@ -561,7 +565,7 @@ def hook_RtlLookupFunctionEntry(ql: Qiling, address: int, params): # If a suitable function entry was found, # compute its address and return. if runtime_function: - return function_table_addr + runtime_function_idx * 12 # sizeof(RUNTIME_FUNCTION) + return function_table_addr + runtime_function_idx * _runtime_function_entry_size(ql, base_addr) return 0 @@ -602,7 +606,7 @@ def hook_RtlLookupFunctionTable(ql: Qiling, address: int, params): function_table = ql.loader.function_tables[base_addr] # compute the total size of the table - size_of_table = len(function_table) * 12 # sizeof(RUNTIME_FUNCTION) + size_of_table = len(function_table) * _runtime_function_entry_size(ql, base_addr) # Write the size to memory at the provided pointer. ql.mem.write_ptr(size_of_table_ptr, size_of_table, 4) diff --git a/qiling/utils.py b/qiling/utils.py index 72e5c32df..9b106a20c 100644 --- a/qiling/utils.py +++ b/qiling/utils.py @@ -235,37 +235,39 @@ def __emu_env_from_macho(path: str) -> Tuple[Optional[QL_ARCH], Optional[QL_OS], def __emu_env_from_pe(path: str) -> Tuple[Optional[QL_ARCH], Optional[QL_OS], Optional[QL_ENDIAN]]: - import pefile + from lief import PE try: - pe = pefile.PE(path, fast_load=True) - except: + pe = PE.parse(path) + except Exception: return None, None, None - arch = None - ostype = None - archendian = None + if pe is None: + return None, None, None + MACHINE_TYPES = PE.Header.MACHINE_TYPES machine_map = { - pefile.MACHINE_TYPE['IMAGE_FILE_MACHINE_I386'] : QL_ARCH.X86, - pefile.MACHINE_TYPE['IMAGE_FILE_MACHINE_AMD64'] : QL_ARCH.X8664, - pefile.MACHINE_TYPE['IMAGE_FILE_MACHINE_ARM'] : QL_ARCH.ARM, - pefile.MACHINE_TYPE['IMAGE_FILE_MACHINE_THUMB'] : QL_ARCH.ARM, - pefile.MACHINE_TYPE['IMAGE_FILE_MACHINE_ARM64'] : QL_ARCH.ARM64 + MACHINE_TYPES.I386 : QL_ARCH.X86, + MACHINE_TYPES.AMD64 : QL_ARCH.X8664, + MACHINE_TYPES.ARM : QL_ARCH.ARM, + MACHINE_TYPES.ARMNT : QL_ARCH.ARM, # Thumb (0x01c2) + MACHINE_TYPES.ARM64 : QL_ARCH.ARM64, } - # get arch - arch = machine_map.get(pe.FILE_HEADER.Machine) + arch = machine_map.get(pe.header.machine) + ostype = None + archendian = None if arch: + SUBSYSTEMS = PE.OptionalHeader.SUBSYSTEM subsystem_uefi = ( - pefile.SUBSYSTEM_TYPE['IMAGE_SUBSYSTEM_EFI_APPLICATION'], - pefile.SUBSYSTEM_TYPE['IMAGE_SUBSYSTEM_EFI_BOOT_SERVICE_DRIVER'], - pefile.SUBSYSTEM_TYPE['IMAGE_SUBSYSTEM_EFI_RUNTIME_DRIVER'], - pefile.SUBSYSTEM_TYPE['IMAGE_SUBSYSTEM_EFI_ROM'] + SUBSYSTEMS.EFI_APPLICATION, + SUBSYSTEMS.EFI_BOOT_SERVICE_DRIVER, + SUBSYSTEMS.EFI_RUNTIME_DRIVER, + SUBSYSTEMS.EFI_ROM, ) - if pe.OPTIONAL_HEADER.Subsystem in subsystem_uefi: + if pe.optional_header.subsystem in subsystem_uefi: ostype = QL_OS.UEFI else: ostype = QL_OS.WINDOWS