avatartwo/avatar2
1
0
Fork 0
mirror of https://github.com/avatartwo/avatar2.git synced 2026-10-12 02:11:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #100 from avatartwo/dev/v1.4.7

Browse Source 
New release for v1.4.7
This commit is contained in:
Marius Muench
2022-02-10 09:30:08 +00:00
committed by GitHub
parent 21ba70e1ae 236ff04f2a
commit 04752c0ac3
26 changed files with 1028 additions and 871 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/build_and_test.yml
-101
View File
@@ -1,101 +0,0 @@
name: Build and test
on:
push:
branches:
- 'main'
- 'dev/*'
pull_request:
branches:
- 'main'
- 'dev/*'
jobs:
build_container:
if: github.repository == 'avatartwo/avatar2'
runs-on: ubuntu-20.04
steps:
- name: Check out repository code
uses: actions/checkout@v2
- name: Generate Dockerfile
run: python3 ${{ github.workspace }}/generate_dockerfile.py --endpoint_list avatar-qemu panda --qemu_targets arm-softmmu mips-softmmu
- name: Build docker container
run: docker build --tag avatar2 .
- name: Test import avatar2
run: docker run --rm avatar2 bash -c "python3 -c 'import avatar2'"
- name: Save and compress the docker image
run: |
docker save avatar2 -o avatar2-image.tar
tar zcvf avatar2-image.tar.gz avatar2-image.tar
- name: Upload docker image
uses: actions/upload-artifact@v2
with:
name: docker-artifact
path: avatar2-image.tar.gz
retention-days: 1
test_cases:
if: github.repository == 'avatartwo/avatar2'
runs-on: ubuntu-20.04
needs: [build_container]
steps:
- name: Check out repository code
uses: actions/checkout@v2
- name: Download docker image
uses: actions/download-artifact@v2
with:
name: docker-artifact
- name: Load docker image
run: |
tar zxvf avatar2-image.tar.gz
docker load --input avatar2-image.tar
docker image ls
- name: Run avatar2 container
run: |
docker run -v ${{ github.workspace }}:/avatar2 --name avatar2 --cap-add=SYS_PTRACE --security-opt seccomp=unconfined -dit avatar2
docker exec avatar2 pip3 install nose
# Test cases
- name: hello_world
run: docker exec avatar2 bash -c 'cd /avatar2/ && python3 ./tests/hello_world.py'
- name: test_remote_memoryprotocol
run: docker exec avatar2 bash -c 'cd /avatar2/ && nosetests ./tests/test_remote_memoryprotocol.py'
- name: test_gdbprotocol
run: docker exec avatar2 bash -c 'cd /avatar2/ && nosetests ./tests/test_gdbprotocol.py'
- name: test_gdbplugin
run: docker exec avatar2 bash -c 'cd /avatar2/ && nosetests ./tests/test_gdbplugin.py'
- name: test_inceptionprotocol
run: docker exec avatar2 bash -c 'cd /avatar2/ && nosetests ./tests/test_inceptionprotocol.py'
- name: gdb_memory_map_loader
run: docker exec avatar2 bash -c 'cd /avatar2/ && python3 ./tests/gdb_memory_map_loader.py'
- name: test_qemutarget ARM
run: docker exec avatar2 bash -c 'cd /avatar2/ && AVATAR2_GDB_EXECUTABLE=gdb-multiarch AVATAR2_ARCH=ARM AVATAR2_QEMU_EXECUTABLE=qemu-system-arm nosetests ./tests/test_qemutarget.py'
- name: test_qemutarget MIPS
run: docker exec avatar2 bash -c 'cd /avatar2/ && AVATAR2_GDB_EXECUTABLE=gdb-multiarch AVATAR2_ARCH=MIPS AVATAR2_QEMU_EXECUTABLE=qemu-system-mips nosetests ./tests/test_qemutarget.py'
- name: test_pyperipheral
run: docker exec avatar2 bash -c 'cd /avatar2/ && AVATAR2_GDB_EXECUTABLE=gdb-multiarch AVATAR2_QEMU_EXECUTABLE=qemu-system-arm nosetests ./tests/pyperipheral/test_pyperipheral.py'
- name: test_qemutarget ARM PANDA
run: docker exec avatar2 bash -c 'cd /avatar2/ && AVATAR2_GDB_EXECUTABLE=gdb-multiarch AVATAR2_ARCH=ARM AVATAR2_QEMU_EXECUTABLE=panda-system-arm nosetests ./tests/test_qemutarget.py'
- name: test_qemutarget MIPS PANDA
run: docker exec avatar2 bash -c 'cd /avatar2/ && AVATAR2_GDB_EXECUTABLE=gdb-multiarch AVATAR2_ARCH=MIPS AVATAR2_QEMU_EXECUTABLE=panda-system-mips nosetests ./tests/test_qemutarget.py'
- name: test_pyperipheral
run: docker exec avatar2 bash -c 'cd /avatar2/ && AVATAR2_GDB_EXECUTABLE=gdb-multiarch AVATAR2_QEMU_EXECUTABLE=panda-system-arm nosetests ./tests/pyperipheral/test_pyperipheral.py'
- name: smoke panda_thumb
run: docker exec avatar2 bash -c 'cd /avatar2/ && AVATAR2_GDB_EXECUTABLE=gdb-multiarch AVATAR2_QEMU_EXECUTABLE=panda-system-arm AVATAR2_PANDA_EXECUTABLE=panda-system-arm nosetests ./tests/smoke/panda_thumb.py'
- name: test_pypandatarget
run: docker exec avatar2 bash -c 'cd /avatar2/ && AVATAR2_GDB_EXECUTABLE=gdb-multiarch AVATAR2_QEMU_EXECUTABLE=panda-system-arm AVATAR2_PANDA_EXECUTABLE=panda-system-arm nosetests ./tests/test_pypandatarget.py'
- name: smoke target_wait
run: docker exec avatar2 bash -c 'cd /avatar2/ && nosetests --processes=-1 --process-timeout=20 ./tests/smoke/target_wait.py'
- name: smoke 54_sync_hooks
run: docker exec avatar2 bash -c 'cd /avatar2/ && AVATAR2_GDB_EXECUTABLE=gdb-multiarch AVATAR2_QEMU_EXECUTABLE=panda-system-arm nosetests --processes=-1 --process-timeout=20 ./tests/smoke/54_sync_hooks.py'
.github/workflows/continuous_integration.yml
+130
View File
@@ -0,0 +1,130 @@
name: CI
on:
push:
branches:
- 'main'
- 'dev**'
pull_request:
branches:
- 'main'
- 'dev**'
jobs:
build:
if: github.repository == 'avatartwo/avatar2'
runs-on: ubuntu-20.04
strategy:
fail-fast: false
matrix:
python-version: ["3.7", "3.8", "3.9"]
steps:
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v2
with:
python-version: ${{ matrix.python-version }}
- name: Install dependencies
run: |
sudo apt-get update
sudo apt-get upgrade -y
sudo apt-get install -y -m -f --install-suggests cmake build-essential git libcapstone3 libcapstone-dev
python -m pip install --upgrade pip
pip install pytest
pip install "cffi>1.14.3" # for pypanda
- name: Check out repository code
uses: actions/checkout@v2
- name: Build avatar2
run: |
cd ${{ github.workspace }}
python setup.py install
- name: Install debuggers targets
run: |
sudo apt-get install -y -m -f --install-suggests gdb gdb-multiarch openocd
- name: Build avatar-qemu target
run: |
cd ${{ github.workspace }}/targets
echo yes | ./build_qemu.sh arm-softmmu,mips-softmmu
cd src/avatar-qemu/build
sudo make install
- name: Run unit tests
env:
AVATAR2_ARCH: ARM
AVATAR2_GDB_EXECUTABLE: gdb-multiarch
run: |
cd ${{ github.workspace }}
python ./tests/hello_world.py
python ./tests/gdb_memory_map_loader.py
- name: Run unit tests for ARM
env:
AVATAR2_ARCH: ARM
AVATAR2_GDB_EXECUTABLE: gdb-multiarch
AVATAR2_QEMU_EXECUTABLE: qemu-system-arm
run: |
cd ${{ github.workspace }}/tests
pytest test_gdbplugin.py \
test_gdbprotocol.py \
test_inceptionprotocol.py \
test_remote_memoryprotocol.py \
test_qemutarget.py
- name: Run unit tests for MIPS
env:
AVATAR2_ARCH: MIPS
AVATAR2_GDB_EXECUTABLE: gdb-multiarch
AVATAR2_QEMU_EXECUTABLE: qemu-system-mips
run: |
cd ${{ github.workspace }}/tests
pytest test_gdbplugin.py \
test_gdbprotocol.py \
test_inceptionprotocol.py \
test_remote_memoryprotocol.py \
test_qemutarget.py
- name: Build panda target
run: |
git clone https://github.com/panda-re/panda.git
cd panda
git checkout 5fb8a5098d0647a13e5e7d39d20ff9f04242572d
sudo prefix=/usr/local ./panda/scripts/install_ubuntu.sh i386-softmmu,arm-softmmu,mips-softmmu
pushd build
sudo make install
# Install pandare in the matrix python environment
sudo chown -R $(id -u):$(id -u) ../panda/python/core
pip install ../panda/python/core
- name: Run unit tests on panda target for ARM
env:
AVATAR2_ARCH: ARM
AVATAR2_GDB_EXECUTABLE: gdb-multiarch
AVATAR2_QEMU_EXECUTABLE: panda-system-arm
AVATAR2_PANDA_EXECUTABLE: panda-system-arm
run: |
cd ${{ github.workspace }}/tests
pytest test_qemutarget.py \
test_pypandatarget.py
- name: Run unit tests on panda target for MIPS
env:
AVATAR2_ARCH: MIPS
AVATAR2_GDB_EXECUTABLE: gdb-multiarch
AVATAR2_QEMU_EXECUTABLE: panda-system-mips
AVATAR2_PANDA_EXECUTABLE: panda-system-mips
run: |
cd ${{ github.workspace }}/tests
pytest test_qemutarget.py \
test_pypandatarget.py
.github/workflows/publish.yml
+31
View File
@@ -0,0 +1,31 @@
name: Publish to PyPI
on:
push:
tags:
- 'v*'
jobs:
build-n-publish:
if: |
github.repository == 'avatartwo/avatar2' && startsWith(github.ref, 'refs/tags/v')
runs-on: ubuntu-20.04
steps:
- name: Check out repository code
uses: actions/checkout@v2
- name: Build a binary wheel and a source tarball
run: python3 setup.py sdist bdist_wheel
- name: Check package
run: |
pip3 install twine
python3 -m twine check dist/*
- name: Publish to PyPI
uses: pypa/gh-action-pypi-publish@release/v1
with:
user: __token__
password: ${{ secrets.PYPI_API_TOKEN }}
.travis.yml
-53
View File
@@ -1,53 +0,0 @@
language: python
python:
- '3.8'
services:
- docker
before_install:
- python3 /home/travis/build/$TRAVIS_REPO_SLUG/generate_dockerfile.py --endpoint_list
avatar-qemu panda --qemu_targets arm-softmmu mips-softmmu
install:
- docker login -u "$DOCKER_USERNAME" -p "$DOCKER_PASSWORD" || true
- docker build --tag avatar2 .
before_script:
- docker run -v /home/travis/build/$TRAVIS_REPO_SLUG:/avatar2 --name avatar2 --cap-add=SYS_PTRACE
--security-opt seccomp=unconfined -dit avatar2
- docker exec avatar2 pip3 install nose
script:
- docker exec avatar2 bash -c 'cd avatar2/ && nosetests ./tests/test_remote_memoryprotocol.py'
- docker exec avatar2 bash -c 'cd avatar2/ && nosetests ./tests/test_gdbprotocol.py'
- docker exec avatar2 bash -c 'cd avatar2/ && nosetests ./tests/test_gdbplugin.py'
- docker exec avatar2 bash -c 'cd avatar2/ && nosetests ./tests/test_inceptionprotocol.py'
- docker exec avatar2 bash -c 'cd avatar2/ && AVATAR2_GDB_EXECUTABLE=gdb-multiarch
AVATAR2_ARCH=ARM AVATAR2_QEMU_EXECUTABLE=qemu-system-arm nosetests ./tests/test_qemutarget.py'
- docker exec avatar2 bash -c 'cd avatar2/ && AVATAR2_GDB_EXECUTABLE=gdb-multiarch
AVATAR2_ARCH=MIPS AVATAR2_QEMU_EXECUTABLE=qemu-system-mips nosetests ./tests/test_qemutarget.py'
- docker exec avatar2 bash -c 'cd avatar2/ && AVATAR2_GDB_EXECUTABLE=gdb-multiarch
AVATAR2_QEMU_EXECUTABLE=qemu-system-arm nosetests ./tests/pyperipheral/test_pyperipheral.py'
- docker exec avatar2 bash -c 'cd avatar2/ && AVATAR2_GDB_EXECUTABLE=gdb-multiarch
AVATAR2_ARCH=ARM AVATAR2_QEMU_EXECUTABLE=panda-system-arm nosetests ./tests/test_qemutarget.py'
- docker exec avatar2 bash -c 'cd avatar2/ && AVATAR2_GDB_EXECUTABLE=gdb-multiarch
AVATAR2_ARCH=MIPS AVATAR2_QEMU_EXECUTABLE=panda-system-mips nosetests ./tests/test_qemutarget.py'
- docker exec avatar2 bash -c 'cd avatar2/ && AVATAR2_GDB_EXECUTABLE=gdb-multiarch
AVATAR2_QEMU_EXECUTABLE=panda-system-arm nosetests ./tests/pyperipheral/test_pyperipheral.py'
- docker exec avatar2 bash -c 'cd avatar2/ && AVATAR2_GDB_EXECUTABLE=gdb-multiarch
AVATAR2_QEMU_EXECUTABLE=panda-system-arm AVATAR2_PANDA_EXECUTABLE=panda-system-arm
nosetests ./tests/smoke/panda_thumb.py'
- docker exec avatar2 bash -c 'cd avatar2/ && AVATAR2_GDB_EXECUTABLE=gdb-multiarch
AVATAR2_QEMU_EXECUTABLE=panda-system-arm AVATAR2_PANDA_EXECUTABLE=panda-system-arm
nosetests ./tests/test_pypandatarget.py'
- docker exec avatar2 bash -c 'cd avatar2/ && python3 ./tests/hello_world.py'
- docker exec avatar2 bash -c 'cd avatar2/ && python3 ./tests/gdb_memory_map_loader.py'
- docker exec avatar2 bash -c 'cd avatar2/ && nosetests --processes=-1 --process-timeout=20
./tests/smoke/target_wait.py'
- docker exec avatar2 bash -c 'cd avatar2/ && AVATAR2_GDB_EXECUTABLE=gdb-multiarch
AVATAR2_QEMU_EXECUTABLE=panda-system-arm nosetests --processes=-1 --process-timeout=20
./tests/smoke/54_sync_hooks.py'
deploy:
provider: pypi
username: "__token__"
password:
secure: IV6/5UW9frUPb51sY4/0VY0DMwhkI46qp1NyjqJwMlWx98ln8xEhdY2SaJNfGLeqTzHYsx5AF9+3PKSB4p3bAIGxkKF7WmS8bs/mjZuObQPN+amskW75QO5InMZJLxBsyjJzI9ljpN6oqGFWFg9TDABp0SpTc9WXNPAejn6Of/oIlpqbjjA5kAavERvHlra4UjzmF7I/l1aL0lxFhdZe0Rnr74lUN4W6A4NikisQtM5HXrB4EOMt6E2FRFlxCYOch7KSELHCqh3cjt8bryFYXW6bEgKoAv2vdkdJeE+3/FpbOMOR0gG/mrkQ0V6yDEeOQNnZBIJLEhovjT7HrXTRxL2yurwQNGNGA7ruw9iCmVKvzFlUTMiE0cmHaLK1ayWIU2hIBr47IZcvucdZmkCSczW9zCHrP4lvsnKILnHtsqfxMKBMwjQhnvlGUTs1Hr13pwbD6uunFFzul1DaYf06rNOnpPW5ZDGqVE4mwTXunMJe47Ko8VPdi4ROjOp92k85RLJp8U+Ye5yBRJxGVNvHTWjEFvu3H5FoPagRxrUxhN1fRWVpl6jELMWoPVMBBMuZgOR17lX0gpl7MECtxpcDFjQ+GXePY+gXLIpn9BpTHUaeFjfoIqMkhVHx9eTKV/LK0GsSUrQDKXsMsqtEvv6CKAorKMzFhoz5mgCkTIU1fdo=
on:
tags: true
CHANGELOG.md
+13 -1
View File
@@ -1,10 +1,22 @@
# Version 1.4.7 (February 10th, 2022)
* Improvements
* Change CI to Github Actions
* avatar-qemu updated to v6.2
* Allow unix sockets for GDBProtocol (thanks @grant-h)
* Bugfixes:
* JLink target and protocol, operational now (thanks @TheSilentDawn)
* pygdbmi version checking (tanks @jcnauta)
* Error messages (thanks @mborgerson and @lacraig2)
# Version 1.4.6 (June 21st, 2021)
* Improvements
* Add build and unit tests to Github Actions
* Add 'ip' alias for ARM register 12
* Bugfizes:
* Bugfixes:
* Unlink RemoteMemory message queue after opened (thanks @grant-h)
* Adjust API to recent pypanda version (0.1.1.2)
* Handle ImportError with RemoteMemoryProtocol on darwin (thanks @fabianfreyer)
README.md
+3 -2
View File
@@ -1,5 +1,5 @@
[![PyPI version](https://badge.fury.io/py/avatar2.svg)](https://badge.fury.io/py/avatar2)
[![Build Status](https://travis-ci.org/avatartwo/avatar2.svg?branch=master)](https://travis-ci.org/avatartwo/avatar2)
[![CI](https://github.com/avatartwo/avatar2/actions/workflows/continuous_integration.yml/badge.svg)](https://github.com/avatartwo/avatar2/actions/workflows/continuous_integration.yml)
<img src="./avatar_logo.png" width="40%" height="40%"><br/>
@@ -107,7 +107,8 @@ The following publications describe, use, or extend the avatar² framework:
- [Paper](https://www.usenix.org/system/files/sec20summer_clements_prepub.pdf) - [Code](https://github.com/embedded-sec/halucinator)
6. C. Cao, L. Guan, J. Ming, P. Liu. "Device-agnostic Firmware Execution is Possible: A Concolic Execution Approach for Peripheral Emulation." Annual Computer Security Applications Conference (ACSAC), December 2020.
- [Paper](https://dl.acm.org/doi/10.1145/3427228.3427280) - [Code](https://github.com/dongmu/Laelaps)
7. G. Hernandez, M. Muench, D. Maier, A. Milburn, S. Park, T. Scharnowski, T. Tucker, P. Traynor, K. R.B. Butler. "FirmWire: Transparent Dynamic Analysis for Cellular Baseband Firmware." Symposium on Network and Distributed System Security (NDSS), April 2022.
- [Paper](https://github.com/FirmWire/FirmWire/blob/main/firmwire-ndss22.pdf?raw=true) - [Code](https://github.com/FirmWire/FirmWire)
# Acknowledgements
Dockerfile → avatar2.Dockerfile
+2 -2
View File
@@ -30,11 +30,11 @@ ARG QEMU_TARGETS="arm-softmmu,mips-softmmu"
RUN sed -i '/deb-src .*-security main restricted/s/^#//g' /etc/apt/sources.list
RUN apt-get update && \
DEBIAN_FRONTEND=noninteractive apt-get build-dep -y qemu && \
apt-get install -y git
apt-get install -y git ninja-build
RUN git clone https://github.com/avatartwo/avatar-qemu /root/avatar-qemu/
RUN cd /root/avatar-qemu/ && \
git checkout dev/qemu-5.1
git checkout dev/qemu-6.2
RUN mkdir -p /root/avatar-qemu/build && cd /root/avatar-qemu/build && \
../configure \
--disable-sdl \
avatar2/protocols/gdb.py
+46 -56
View File
@@ -1,4 +1,5 @@
import sys
from pkg_resources import packaging
from threading import Thread, Event, Condition
from struct import pack, unpack
from codecs import encode
@@ -267,8 +268,8 @@ class GDBProtocol(object):
if local_arguments is not None:
gdb_args += [local_arguments]
if sys.version_info <= (3, 6):
# See breaking change in pygdbmi: https://github.com/cs01/pygdbmi/releases
if packaging.version.parse(pygdbmi.__version__) < packaging.version.parse("0.10.0.0"):
self._gdbmi = pygdbmi.gdbcontroller.GdbController(
gdb_path=gdb_executable,
gdb_args=gdb_args,
@@ -293,10 +294,10 @@ class GDBProtocol(object):
self.shutdown()
def shutdown(self):
if self._communicator is not None:
if hasattr(self, "_communicator") and self._communicator is not None:
self._communicator.stop()
self._communicator = None
if self._gdbmi is not None:
if hasattr(self, "_gdbmi") and self._gdbmi is not None:
self._gdbmi.exit()
self._gdbmi = None
@@ -336,15 +337,7 @@ class GDBProtocol(object):
return ret
def remote_connect(self, ip='127.0.0.1', port=3333):
"""
connect to a remote gdb server
:param ip: ip of the remote gdb-server (default: localhost)
:param port: port of the remote gdb-server (default: port)
:returns: True on successful connection
"""
def _remote_connect_common(self, remote_string, transport_setup=None):
req = ['-gdb-set', 'target-async', 'on']
ret, resp = self._sync_request(req, GDB_PROT_DONE)
if not ret:
@@ -355,8 +348,6 @@ class GDBProtocol(object):
req = ['-gdb-set', 'architecture', self._arch.gdb_name]
ret, resp = self._sync_request(req, GDB_PROT_DONE)
if not ret:
self.log.critical(
"Unable to set architecture, received response: %s" %
@@ -378,7 +369,11 @@ class GDBProtocol(object):
resp)
raise Exception("GDBProtocol was unable to set endianness")
req = ['-target-select', 'remote', '%s:%d' % (ip, int(port))]
# transport unique setup, if applicable
if transport_setup:
transport_setup()
req = ['-target-select', 'remote', remote_string]
ret, resp = self._sync_request(req, GDB_PROT_CONN)
self.log.debug(
@@ -392,6 +387,26 @@ class GDBProtocol(object):
return ret
def remote_connect(self, ip='127.0.0.1', port=3333):
"""
connect to a remote gdb server via TCP
:param ip: ip of the remote gdb-server (default: localhost)
:param port: port of the remote gdb-server (default: port)
:returns: True on successful connection
"""
return self._remote_connect_common('%s:%d' % (ip, int(port)))
def remote_connect_unix(self, unix_path):
"""
connect to a remote gdb server via a UNIX domain socket path
:param unix_path: path of the UNIX domain socket
:returns: True on successful connection
"""
return self._remote_connect_common(unix_path)
def remote_connect_serial(self, device='/dev/ttyACM0', baud_rate=38400,
parity='none'):
"""
@@ -403,49 +418,24 @@ class GDBProtocol(object):
:returns: True on successful connection
"""
req = ['-gdb-set', 'architecture', self._arch.gdb_name]
ret, resp = self._sync_request(req, GDB_PROT_DONE)
if not ret:
self.log.critical(
"Unable to set architecture, received response: %s" %
resp)
raise Exception(("GDBProtocol was unable to set the architecture\n"
"Did you select the right gdb_executable?"))
def serial_setup():
if parity not in ['none', 'even', 'odd']:
self.log.critical("Parity must be none, even or odd")
raise Exception("Cannot set parity to %s" % parity)
if parity not in ['none', 'even', 'odd']:
self.log.critical("Parity must be none, even or odd")
raise Exception("Cannot set parity to %s" % parity)
req = ['-gdb-set', 'serial', 'parity', '%s' % parity]
ret, resp = self._sync_request(req, GDB_PROT_DONE)
if not ret:
self.log.critical("Unable to set parity")
raise Exception("GDBProtocol was unable to set parity")
req = ['-gdb-set', 'mi-async', 'on']
ret, resp = self._sync_request(req, GDB_PROT_DONE)
if not ret:
self.log.critical(
"Unable to set GDB/MI to async, received response: %s" %
resp)
raise Exception("GDBProtocol was unable to connect")
req = ['-gdb-set', 'serial', 'baud', '%i' % baud_rate]
ret, resp = self._sync_request(req, GDB_PROT_DONE)
if not ret:
self.log.critical("Unable to set baud rate")
raise Exception("GDBProtocol was unable to set Baudrate")
req = ['-gdb-set', 'serial', 'parity', '%s' % parity]
ret, resp = self._sync_request(req, GDB_PROT_DONE)
if not ret:
self.log.critical("Unable to set parity")
raise Exception("GDBProtocol was unable to set parity")
req = ['-gdb-set', 'serial', 'baud', '%i' % baud_rate]
ret, resp = self._sync_request(req, GDB_PROT_DONE)
if not ret:
self.log.critical("Unable to set baud rate")
raise Exception("GDBProtocol was unable to set Baudrate")
req = ['-target-select', 'remote', '%s' % device]
ret, resp = self._sync_request(req, GDB_PROT_CONN)
self.log.debug(
"Attempted to connect to target. Received response: %s" %
resp)
self.update_target_regs()
return ret
return self._remote_connect_common(device, transport_setup=serial_setup)
def remote_disconnect(self):
"""
avatar2/protocols/jlink.py
+69 -29
View File
@@ -1,4 +1,6 @@
import sys
import string
import binascii
import pylink
from time import sleep
from threading import Thread, Event, Condition
@@ -24,12 +26,14 @@ class JLinkProtocol(Thread):
:ivar origin: the target utilizing this protocol
"""
def __init__(self, serial="12345678", device="ARM7", avatar=None, origin=None):
def __init__(self, serial=None, device="ARM7", interface="swd", avatar=None, origin=None):
self._shutdown = Event()
self.avatar = avatar
self.origin = origin
self._origin = origin
self.jlink = pylink.JLink()
self.jlink.open(serial)
self.jlink.open(serial_no=serial)
if interface == "swd": # swd is more generic than jtag
self.jlink.set_tif(pylink.enums.JLinkInterfaces.SWD)
self.log = logging.getLogger('%s.%s' %
(origin.log.name, self.__class__.__name__)
) if origin else \
@@ -71,8 +75,8 @@ class JLinkProtocol(Thread):
name = self.jlink.register_name(idx)
regs[name] = idx
if hasattr(self.origin, 'regs'):
self.origin.regs._update(regs)
if hasattr(self._origin, 'regs'):
self._origin.regs._update(regs)
def run(self):
# Target state management thread
@@ -84,25 +88,25 @@ class JLinkProtocol(Thread):
try:
while not self._shutdown.is_set():
is_halted = self.jlink.halted()
if is_halted and self.origin.state == TargetStates.RUNNING:
if is_halted and self._origin.state == TargetStates.RUNNING:
# We just halted
# But did we hit a BP?
self.log.debug("JLink Target is halting...")
avatar_msg = UpdateStateMessage(self.origin, TargetStates.STOPPED)
avatar_msg = UpdateStateMessage(self._origin, TargetStates.STOPPED)
self.avatar.fast_queue.put(avatar_msg)
self.origin.wait()
self._origin.wait()
self.log.debug("JLink target has halted")
pc = self.get_pc()
if self.jlink.breakpoint_find(pc):
self.log.debug("JLink Target hit breakpoint %d" % self.jlink.breakpoint_find(pc))
avatar_msg = BreakpointHitMessage(self.origin, self.jlink.breakpoint_find(pc), pc)
avatar_msg = BreakpointHitMessage(self._origin, self.jlink.breakpoint_find(pc), pc)
self.avatar.queue.put(avatar_msg)
elif not is_halted and self.origin.state == TargetStates.STOPPED:
self.log.debug("About to resume target.")
avatar_msg = UpdateStateMessage(self.origin, TargetStates.RUNNING)
elif not is_halted and self._origin.state == TargetStates.STOPPED:
self.log.info("About to resume target.")
avatar_msg = UpdateStateMessage(self._origin, TargetStates.RUNNING)
self.avatar.fast_queue.put(avatar_msg)
while self.origin.state != TargetStates.RUNNING:
while self._origin.state != TargetStates.RUNNING:
pass
self.log.debug("JLink target has resumed")
except:
@@ -147,7 +151,7 @@ class JLinkProtocol(Thread):
"""Writes memory
:param address: Address to write to
:param wordsize: the size of the write (1, 2, 4 or 8)
:param wordsize: the size of the write (1, 2, 4)
:param val: the written value
:type val: int if num_words == 1 and raw == False
list if num_words > 1 and raw == False
@@ -162,37 +166,66 @@ class JLinkProtocol(Thread):
raise ValueError("val had zero length")
new_val = [ord(v) for v in val]
val = new_val
if not isinstance(val, list):
val = [val]
try:
self.jlink.memory_write(address, contents)
self.jlink.memory_write(address, data=val, nbits=wordsize * 8)
return True
except pylink.JLinkException:
return False
def read_memory(self, address, wordsize=4, num_words=1, raw=False):
def read_memory(self, address, wordsize=1, num_words=1, raw=False):
"""reads memory
:param address: Address to write to
:param wordsize: the size of a read word (1, 2, 4 or 8)
:param wordsize: the size of a read word (1, 2, 4).
nbits if provided, must be either 8, 16, or 32.
If not provided, always reads num_units bytes.
Ref https://pylink.readthedocs.io/en/latest/pylink.html
:param num_words: the amount of read words
:param raw: Whether the read memory should be returned unprocessed
:return: The read memory
"""
ret = self.jlink.memory_read(address, num_units=num_words, nbits=wordsize)
ret = self.jlink.memory_read(address, num_units=num_words, nbits=wordsize * 8) # nbits indicate bits of each unit
if raw:
raw_mem = "".join([newint.to_bytes(i, length=int(math.ceil(i.bit_length() / 8.0))) for i in ret])
raw_mem = b''
for i in ret:
raw_mem += i.to_bytes(wordsize, "little")
return raw_mem
return ret
if num_words==1:
return ret[0]
else:
return ret
def read_register(self, reg):
the_reg = tolower(reg)
"""read_register
jlink supported registers: ['R0', 'R1', 'R2', 'R3', 'R4', 'R5', 'R6', 'R7', 'R8', 'R9', 'R10', 'R11', 'R12', 'R13 (SP)', 'R14', 'R15 (PC)', 'XPSR', 'MSP', 'PSP', 'APSR', 'EPSR', 'IPSR', 'PRIMASK', 'BASEPRI', 'FAULTMASK', 'CONTROL', 'BASEPRI_MAX', 'IAPSR', 'EAPSR', 'IEPSR', 'FPSCR', 'FPS0', 'FPS1', 'FPS2', 'FPS3', 'FPS4', 'FPS5', 'FPS6', 'FPS7', 'FPS8', 'FPS9', 'FPS10', 'FPS11', 'FPS12', 'FPS13', 'FPS14', 'FPS15', 'FPS16', 'FPS17', 'FPS18', 'FPS19', 'FPS20', 'FPS21', 'FPS22', 'FPS23', 'FPS24', 'FPS25', 'FPS26', 'FPS27', 'FPS28', 'FPS29', 'FPS30', 'FPS31', 'CycleCnt', 'MSP_NS', 'PSP_NS', 'MSP_S', 'PSP_S', 'MSPLIM_S', 'PSPLIM_S', 'MSPLIM_NS', 'PSPLIM_NS', 'CFBP_S', 'CFBP_NS', 'PRIMASK_NS', 'BASEPRI_NS', 'FAULTMASK_NS', 'CONTROL_NS', 'BASEPRI_MAX_NS', 'PRIMASK_S', 'BASEPRI_S', 'FAULTMASK_S', 'CONTROL_S', 'BASEPRI_MAX_S', 'MSPLIM', 'PSPLIM', 'BASEPRI_BASE0', 'FAULTMASK_BASE0', 'CONTROL_BASE0', 'BASEPRI_MAX_BASE0']
:param reg: register name string
:return: register value
"""
the_reg = reg.upper() # all register names are upper, not lower
# jlink has no reg named ip, lr, cpsr. So transfer them
if the_reg == 'IP':
the_reg = 'R12'
if the_reg == 'LR':
the_reg = 'R14'
if the_reg == 'CPSR':
the_reg = 'XPSR'
the_idx = -1
for idx in self.jlink.register_list():
if the_reg == self.jlink.register_name(idx):
the_idx = idx
break
return self.register_read(the_idx)
if(idx == 13 or idx == 15): # R13 (SP) and R15 (PC) need special operation
if the_reg in self.jlink.register_name(idx):
the_idx = idx
break
else:
if the_reg == self.jlink.register_name(idx):
the_idx = idx
break
if(the_idx == -1):
self.log.exception("Do not find target register")
return self.jlink.register_read(the_idx)
def get_pc(self):
# Get PC a shitty way
@@ -203,12 +236,19 @@ class JLinkProtocol(Thread):
def write_register(self, reg, val):
"""Set one register on the target
:returns: True on success"""
the_reg = tolower(reg)
the_reg = reg.upper()
the_idx = -1
for idx in self.jlink.register_list():
if the_reg == self.jlink.register_name(idx):
the_idx = idx
break
if(idx == 13 or idx == 15): # R13 (SP) and R15 (PC) need special operation
if the_reg in self.jlink.register_name(idx):
the_idx = idx
break
else:
if the_reg == self.jlink.register_name(idx):
the_idx = idx
break
if(the_idx == -1):
self.log.exception("Do not find target register")
return self.jlink.register_write(the_idx, val)
def step(self):
avatar2/targets/gdb_target.py
+8 -1
View File
@@ -8,6 +8,7 @@ class GDBTarget(Target):
def __init__(self, avatar,
gdb_executable=None, gdb_additional_args=None,
gdb_ip='127.0.0.1', gdb_port=3333,
gdb_unix_socket_path=None,
gdb_serial_device='/dev/ttyACM0',
gdb_serial_baud_rate=38400,
gdb_serial_parity='none',
@@ -26,6 +27,7 @@ class GDBTarget(Target):
self.gdb_additional_args = gdb_additional_args if gdb_additional_args else []
self.gdb_ip = gdb_ip
self.gdb_port = gdb_port
self.gdb_unix_socket_path = gdb_unix_socket_path
self.gdb_serial_device = gdb_serial_device
self.gdb_serial_baud_rate = gdb_serial_baud_rate
self.gdb_serial_parity = gdb_serial_parity
@@ -49,7 +51,12 @@ class GDBTarget(Target):
# If we are debugging a program locally,
# we do not need to establish any connections
if not self._local_binary:
if not self._serial:
if self.gdb_unix_socket_path is not None:
if gdb.remote_connect_unix(self.gdb_unix_socket_path):
self.log.info("Connected to Target")
else:
self.log.warning("Connecting failed")
elif not self._serial:
if gdb.remote_connect(ip=self.gdb_ip, port=self.gdb_port):
self.log.info("Connected to Target")
else:
avatar2/targets/jlink_target.py
+8 -3
View File
@@ -10,7 +10,7 @@ else:
class JLinkTarget(Target):
def __init__(self, avatar, serial, device, **kwargs):
def __init__(self, avatar, serial=None, device="ARM7", interface="swd", **kwargs):
"""
Create a JLink target instance
:param avatar: The avatar instance
@@ -21,11 +21,12 @@ class JLinkTarget(Target):
super(JLinkTarget, self).__init__(avatar, **kwargs)
self.avatar = avatar
self.serial = serial
self.interface = interface
self.device = device
@watch("TargetInit")
def init(self):
jlink = JLinkProtocol(serial=self.serial, device=self.device, avatar=self.avatar, origin=self)
jlink = JLinkProtocol(serial=self.serial, device=self.device, interface=self.interface, avatar=self.avatar, origin=self)
self.protocols.set_all(jlink)
if jlink.jlink.halted():
self.state = TargetStates.STOPPED
@@ -34,4 +35,8 @@ class JLinkTarget(Target):
#self.wait()
def reset(self, halt=True):
self.protocols.execution.reset(halt=halt)
self.protocols.execution.reset(halt=halt)
if halt:
self.state = TargetStates.STOPPED
else:
self.state = TargetStates.RUNNING
avatar2/targets/pypanda_target.py
+1 -1
View File
@@ -18,7 +18,7 @@ class PyPandaTarget(PandaTarget):
except ImportError:
raise RuntimeError(("PyPanda could not be found! for installation, "
"please follow the steps at https://github.com/"
"panda-re/panda/blob/master/panda/pypanda/docs/USAGE.md"))
"panda-re/panda/blob/dev/panda/python/docs/USAGE.md"))
kwargs['executable'] = abspath(pandare.__file__)
super(PyPandaTarget, self).__init__(*args, **kwargs)
avatar2/targets/qemu_target.py
+17 -2
View File
@@ -25,6 +25,7 @@ class QemuTarget(Target):
firmware=None,
gdb_executable=None,
gdb_port=3333,
gdb_unix_socket_path=None,
additional_args=None,
gdb_additional_args=None,
gdb_verbose=False,
@@ -58,6 +59,7 @@ class QemuTarget(Target):
)
self.gdb_port = gdb_port
self.gdb_unix_socket_path = gdb_unix_socket_path
self.gdb_additional_args = gdb_additional_args if gdb_additional_args else []
self.gdb_verbose = gdb_verbose
@@ -90,7 +92,10 @@ class QemuTarget(Target):
machine = ["-machine", "configurable"]
kernel = ["-kernel", self.qemu_config_file]
gdb_option = ["-gdb", "tcp::" + str(self.gdb_port)]
if self.gdb_unix_socket_path:
gdb_option = ["-gdb", "unix:%s,server,nowait" % str(self.gdb_unix_socket_path)]
else:
gdb_option = ["-gdb", "tcp::" + str(self.gdb_port)]
stop_on_startup = ["-S"]
nographic = ["-nographic"] # , "-monitor", "/dev/null"]
qmp = ["-qmp", "tcp:127.0.0.1:%d,server,nowait" % self.qmp_port]
@@ -301,10 +306,20 @@ class QemuTarget(Target):
self.protocols.monitor = qmp
self.protocols.remote_memory = rmp
if gdb.remote_connect(port=self.gdb_port) and qmp.connect():
connect_success = True
if self.gdb_unix_socket_path:
connect_success = connect_success and gdb.remote_connect_unix(self.gdb_unix_socket_path)
else:
connect_success = connect_success and gdb.remote_connect(port=self.gdb_port)
connect_success = connect_success and qmp.connect()
if connect_success:
self.log.info("Connected to remote target")
else:
self.log.warning("Connection to remote target failed")
if rmp:
rmp.connect()
self.wait()
generate_dockerfile.py
+2 -2
View File
@@ -53,11 +53,11 @@ TEMPLATE_QEMU_BUILD="""
RUN sed -i '/deb-src .*-security main restricted/s/^#//g' /etc/apt/sources.list
RUN apt-get update && \\
DEBIAN_FRONTEND=noninteractive apt-get build-dep -y qemu && \\
apt-get install -y git
apt-get install -y git ninja-build
RUN git clone https://github.com/avatartwo/avatar-qemu /root/avatar-qemu/
RUN cd /root/avatar-qemu/ && \\
git checkout dev/qemu-5.1
git checkout dev/qemu-6.2
RUN mkdir -p /root/avatar-qemu/build && cd /root/avatar-qemu/build && \\
../configure \\
--disable-sdl \\
handbook/0x02_targets.md
+10
View File
@@ -100,6 +100,16 @@ parameters can be specified on a OpenOCDTarget:
| gdb_additional_args | [str] | [] | List with additional arguments which shall be passed to gdb |
| gdb_port | int | 3333 | Port on which the gdbserver being connected to listens |
## JLinkTarget
The JLinkTarget provides similar functions as OpenOCDTarget. It could use SWD and JTAG interface to control the physical devices.
Easy to use. For example, with Segger JLink hardware devices, developers do not need to prepare the configuration files which are needed by OpenOCD for each devices. The following parameters can be specified on a JLinkTarget:
| name | type | default | purpose |
|---------------------|-------|---------|-------------------------------------------------------------------------------------------------|
| serial | str | None | JLink emulator serial number. Default None, which leaves JLink the choice where to connect to. |
| device | str | "ARM7" | Device name which should be avaliable for used JLink emulator |
| interface | str | "swd" | "swd" or "jtag" interface |
## QemuTarget
Qemu is a full-system emulator which has beed modified for avatar² in order to
allow a complete free (hardware-) configuration of the system to be emulated and
targets/build_panda.sh
+32 -44
View File
@@ -1,51 +1,39 @@
#!/bin/bash
source /etc/os-release
# Usage ./build_panda.sh [architectures]
# example: ./build_panda.sh arm-softmmu,mips-softmmu
#
repo="deb-src http://archive.ubuntu.com/ubuntu/ $UBUNTU_CODENAME-security main restricted"
apt_src="/etc/apt/sources.list"
TARGET_LIST="arm-softmmu,mips-softmmu,i386-softmmu"
PANDA_NPROC=${PANDA_NPROC:-$(nproc)}
set -e
if [[ "$ID" == "ubuntu" ]]
set -ex
source /etc/os-release
if [[ "$ID" != "ubuntu" ]]
then
if ! grep -q '^deb-src .*'$UBUNTU_CODENAME'-security main restricted' $apt_src;
then
echo "[WARNING] This script is about to add '$repo' to $apt_src"
read -p "Do you want to continue? (Yes/No) " cont
case $cont in
Yes|yes|Y|y )
sudo bash -c "echo '$repo' >> $apt_src"
echo "Continuing installation..."
;;
* ) echo "Aborting..."
exit -1
;;
esac
fi
# Use the panda install_ubuntu.sh script
cd `dirname "$BASH_SOURCE"`/src/
git submodule update --init avatar-panda
cd ..
src/avatar-panda/panda/scripts/install_ubuntu.sh --disable-sdl --target-list=arm-softmmu
mkdir -p build/
mv src/avatar-panda/build/ build/panda/
else
echo "[Warning] Attempting to run installation on a non-ubuntu system."
echo "You may have to install dependencies manually"
cd `dirname "$BASH_SOURCE"`/src/
git submodule update --init avatar-panda
cd avatar-panda
git submodule update --init dtc
mkdir -p ../../build/panda/panda
cd ../../build/panda/panda
../../../src/avatar-panda/configure --disable-sdl --target-list=arm-softmmu
make -j ${PANDA_NPROC}
echo "[ERROR] This script only supports Ubuntu systems"
exit -1
fi
if [[ $# -ge 1 ]]
then
TARGET_LIST="$1"
fi
echo "Building for targets: $TARGET_LIST"
echo ""
# Use the panda install_ubuntu.sh script
cd `dirname "$BASH_SOURCE"`/src/
#git submodule update --init avatar-panda
cd avatar-panda
./panda/scripts/install_ubuntu.sh \
--disable-sdl \
--target-list=$TARGET_LIST
echo ""
echo "Export the following env variable:"
echo "export AVATAR2_PANDA_EXECUTABLE=$PWD/arm-softmmu/panda-system-arm"
targets/build_qemu.sh
+37 -11
View File
@@ -1,19 +1,30 @@
#!/bin/bash
# Usage ./build_qemu.sh [architectures]
# example: ./build_qemu.sh arm-softmmu,mips-softmmu
#
source /etc/os-release
repo="deb-src http://archive.ubuntu.com/ubuntu/ $UBUNTU_CODENAME-security main restricted"
apt_src="/etc/apt/sources.list"
QEMU_NPROC=$(QEMU_NPROC:-$(nproc))
TARGET_LIST="arm-softmmu,mips-softmmu"
REPO="deb-src http://archive.ubuntu.com/ubuntu/ $UBUNTU_CODENAME-security main restricted"
APT_SRC="/etc/apt/sources.list"
QEMU_NPROC=${QEMU_NPROC:-$(nproc)}
set -ex
if [[ "$ID" == "ubuntu" ]]
then
if ! grep -q '^deb-src .*'$UBUNTU_CODENAME'-security main restricted' $apt_src;
if [ $EUID -ne 0 ]; then
SUDO=sudo
fi
if ! grep -q '^deb-src .*'$UBUNTU_CODENAME'-security main restricted' $APT_SRC;
then
echo "[WARNING] This script is about to add '$repo' to $apt_src"
echo "[WARNING] This script is about to add '$REPO' to $APT_SRC"
read -p "Do you want to continue? (Yes/No) " cont
case $cont in
Yes|yes|Y|y )
sudo bash -c "echo '$repo' >> $apt_src"
$SUDO bash -c "echo '$REPO' >> $APT_SRC"
echo "Continuing installation..."
;;
* ) echo "Aborting..."
@@ -21,22 +32,37 @@ then
;;
esac
fi
sudo apt-get update
sudo apt-get build-dep -y qemu
$SUDO apt-get update
DEBIAN_FRONTEND=noninteractive $SUDO apt-get build-dep -y qemu
$SUDO apt-get install -y git ninja-build
else
echo "[Warning] Attempting to run installation on a non-ubuntu system."
echo "You may have to install dependencies manually"
fi
if [[ $# -ge 1 ]]
then
TARGET_LIST="$1"
fi
echo "Building for targets: $TARGET_LIST"
cd `dirname "$BASH_SOURCE"`/src/
git submodule update --init avatar-qemu
cd avatar-qemu
git submodule update --init dtc
mkdir -p ../../build/qemu/
cd ../../build/qemu
../../src/avatar-qemu/configure --disable-sdl --target-list=arm-softmmu
mkdir -p build
cd build
../configure \
--disable-sdl \
--target-list=$TARGET_LIST
make -j $QEMU_NPROC
echo ""
echo "Export the following env variable:"
echo "export AVATAR2_PANDA_EXECUTABLE=$PWD/arm-softmmu/qemu-system-arm"
targets/src/avatar-panda
+1 -1
Submodule targets/src/avatar-panda updated: 1b26d2ba21...5fb8a5098d
targets/src/avatar-qemu
+1 -1
Submodule targets/src/avatar-qemu updated: ee8fcc7449...d774496465
tests/test_gdbplugin.py
+109 -103
View File
@@ -1,3 +1,5 @@
import unittest
from avatar2.protocols.gdb import GDBProtocol
import avatar2
@@ -7,138 +9,142 @@ import time
import sys
from os.path import dirname, abspath
from os.path import dirname, abspath, realpath
from nose.tools import *
SLEEP_TIME = 2
MEM_ADDR = 0x555555554000
MAIN_ADDR = 0x5555555546b4
SLEEP_TIME = .1
# TODO: Resolve those dynamically.
# They can change based on the environment. Unfortunately, for now
# The GDB stub cannot resolve them
MEM_ADDR = 0x555555400000
MAIN_ADDR = 0x5555554006b4
XML_PATH = dirname(abspath((__file__))) \
+ '/../avatar2/plugins/gdb/x86_64-target.xml'
port = 4444
avatar_gdb_port = 7777
a = None
s = None
p = None
g = None
x86_regs = [u'rax', u'rbx', u'rcx', u'rdx', u'rsi', u'rdi', u'rbp', u'rsp',
PORT = 4444
AV_GDB_PORT = 7777
X86_REGS = [u'rax', u'rbx', u'rcx', u'rdx', u'rsi', u'rdi', u'rbp', u'rsp',
u'r8', u'r9', u'r10', u'r11', u'r12', u'r13', u'r14', u'r15',
u'rip', u'eflags', u'cs', u'ss', u'ds', u'es', u'fs', u'gs']
def setup_avatar_gdb_server():
global a,s
a = avatar2.Avatar(arch=avatar2.archs.X86_64)
gdb_target = a.add_target(avatar2.GDBTarget, gdb_port=avatar_gdb_port)
a.init_targets()
a.load_plugin('gdbserver')
s = a.spawn_gdb_server(gdb_target, port, True, XML_PATH)
class GdbPluginTestCase(unittest.TestCase):
def setUp(self):
pass
def setup_avatar_gdb_server(self):
self.avatar = avatar2.Avatar(arch=avatar2.archs.X86_64)
self.gdb_target = self.avatar.add_target(avatar2.GDBTarget, gdb_port=AV_GDB_PORT)
self.avatar.init_targets()
self.avatar.load_plugin('gdbserver')
self.sk = self.avatar.spawn_gdb_server(self.gdb_target, PORT, True, XML_PATH)
def setup_env(self, binary):
self.process = subprocess.Popen(['gdbserver', '--once', '127.0.0.1:%d' %
AV_GDB_PORT, binary],
stderr=subprocess.PIPE)
out = str(self.process.stderr.readline())
self.assertEqual(binary in out, True, out)
out = str(self.process.stderr.readline())
self.assertEqual(str(AV_GDB_PORT) in out, True, out)
# create avatar instance offering the gdbserver
self.setup_avatar_gdb_server()
self.gdb = GDBProtocol(arch=avatar2.archs.X86_64)
self.gdb.remote_connect(port=PORT)
def wait_stopped(self):
# As we do not have access to avatar synchronizing target states
# on this level, we apply this little hack to synchronize the target
while True:
ret, out = self.gdb.console_command('info program')
if 'Program stopped' in out:
break
time.sleep(SLEEP_TIME)
def tearDown(self):
self.sk.shutdown()
self.avatar.shutdown()
self.gdb.shutdown()
self.process.terminate()
def setup_helloworld():
global p, g, port
class TestCaseOnHelloWorld(GdbPluginTestCase):
def setUp(self):
dir_path = dirname(realpath(__file__))
binary = '%s/binaries/hello_world' % dir_path
self.setup_env(binary)
def test_register_names(self):
regs = self.gdb.get_register_names()
self.assertListEqual(regs[:len(X86_REGS)], X86_REGS, regs)
binary = '%s/tests/binaries/hello_world' % os.getcwd()
p = subprocess.Popen(['gdbserver', '--once', '127.0.0.1:%d' %
avatar_gdb_port, binary],
stderr=subprocess.PIPE)
out = str(p.stderr.readline())
assert_equal(binary in out, True)
out = str(p.stderr.readline())
assert_equal(str(avatar_gdb_port) in out, True)
# create avatar instance offering the gdbserver
setup_avatar_gdb_server()
def test_register_read_and_write(self):
g = GDBProtocol(arch=avatar2.archs.X86_64)
g.remote_connect(port=port)
ret = self.gdb.write_register('rbx', 1678)
ret = self.gdb.read_register('rbx')
self.assertEqual(ret, 1678, ret)
def setup_inf_loop():
global p, g, port
def test_break_run_and_read_write_mem(self):
binary = '%s/tests/binaries/infinite_loop' % os.getcwd()
p = subprocess.Popen(['gdbserver', '--once', '127.0.0.1:%d' %
avatar_gdb_port, binary],
stderr=subprocess.PIPE)
ret = self.gdb.set_breakpoint(MAIN_ADDR) # we don't support execfile transfer yet
self.assertEqual(ret, True, ret)
out = str(p.stderr.readline())
assert_equal(binary in out, True)
out = str(p.stderr.readline())
assert_equal(str(avatar_gdb_port) in out, True)
ret = self.gdb.cont()
self.assertEqual(ret, True, ret)
setup_avatar_gdb_server()
g = GDBProtocol(arch=avatar2.archs.X86_64)
g.remote_connect(port=port)
self.wait_stopped()
ret = self.gdb.read_memory(MEM_ADDR, 4)
self.assertEqual(ret, 0x464c457f, ret)
ret = self.gdb.write_memory(MEM_ADDR, 4, 0x41414141)
self.assertEqual(ret, True, ret)
ret = self.gdb.read_memory(MEM_ADDR, 4)
self.assertEqual(ret, 0x41414141, ret)
def teardown_func():
s.shutdown()
a.shutdown()
g.shutdown()
p.terminate()
class TestCaseOnInfiniteLoop(GdbPluginTestCase):
@with_setup(setup_helloworld, teardown_func)
def test_register_names():
regs = g.get_register_names()
assert_list_equal(regs[:len(x86_regs)], x86_regs)
def setUp(self):
dir_path = dirname(realpath(__file__))
binary = '%s/binaries/infinite_loop' % dir_path
self.setup_env(binary)
@with_setup(setup_helloworld, teardown_func)
def test_register_read_and_write():
def test_continue_stopping_stepping(self):
ret = g.write_register('rbx', 1678)
ret = g.read_register('rbx')
assert_equal(ret, 1678)
ret = self.gdb.cont()
self.assertEqual(ret, True, ret)
ret = self.gdb.stop()
self.assertEqual(ret, True, ret)
self.wait_stopped()
ret = self.gdb.step()
self.assertEqual(ret, True, ret)
self.wait_stopped()
@with_setup(setup_helloworld, teardown_func)
def test_break_run_and_read_write_mem():
ret = g.set_breakpoint(MAIN_ADDR) # we don't support execfile transfer yet
assert_equal(ret, True)
ret = g.cont()
assert_equal(ret, True)
# todo: enable waiting
time.sleep(SLEEP_TIME)
ret = g.read_memory(MEM_ADDR, 4)
assert_equal(ret, 0x464c457f)
ret = g.write_memory(MEM_ADDR, 4, 0x41414141)
assert_equal(ret, True)
ret = g.read_memory(MEM_ADDR, 4)
assert_equal(ret, 0x41414141)
@with_setup(setup_inf_loop, teardown_func)
def test_continue_stopping_stepping():
ret = g.cont()
assert_equal(ret, True)
time.sleep(SLEEP_TIME)
ret = g.stop()
assert_equal(ret, True)
time.sleep(SLEEP_TIME)
ret = g.step()
assert_equal(ret, True)
if __name__ == '__main__':
setup_inf_loop()
test_continue_stopping_stepping()
teardown_func()
unittest.main()
tests/test_gdbprotocol.py
+122 -95
View File
@@ -1,131 +1,158 @@
from avatar2.protocols.gdb import GDBProtocol
import avatar2
import unittest
import subprocess
import os
import time
import re
from nose.tools import *
from os.path import dirname, realpath
SLEEP_TIME = 1
import avatar2
from avatar2.protocols.gdb import GDBProtocol
MEM_ADDR = 0x555555554000
port = 4444
p = None
g = None
from utils import unix2tcp
x86_regs = [u'rax', u'rbx', u'rcx', u'rdx', u'rsi', u'rdi', u'rbp', u'rsp',
SLEEP_TIME = .1
PORT = 4444
X86_REGS = [u'rax', u'rbx', u'rcx', u'rdx', u'rsi', u'rdi', u'rbp', u'rsp',
u'r8', u'r9', u'r10', u'r11', u'r12', u'r13', u'r14', u'r15',
u'rip', u'eflags', u'cs', u'ss', u'ds', u'es', u'fs', u'gs']
def setup_helloworld():
global p, g, port
class GdbProtocolTestCase(unittest.TestCase):
def setUp(self):
pass
def setup_env(self, binary, unix_socket=False):
self.process = subprocess.Popen(['gdbserver', '--once', '127.0.0.1:%d' % PORT, binary],
stderr=subprocess.PIPE)
out = str(self.process.stderr.readline())
self.assertEqual(binary in out, True, out)
out = str(self.process.stderr.readline())
self.assertEqual(str(PORT) in out, True, out)
self.gdb = GDBProtocol(arch=avatar2.archs.X86_64)
if unix_socket is True:
socket_path ='/tmp/test_socket'
unix2tcp(socket_path, "127.0.0.1", PORT)
self.gdb.remote_connect_unix(socket_path)
else:
self.gdb.remote_connect(port=PORT)
# Base addresses can change across kernel versions due to PIE binaries
self.base_address = self.gdb.get_symbol("main")[1] & ~0xfff
binary = '%s/tests/binaries/hello_world' % os.getcwd()
p = subprocess.Popen(['gdbserver', '--once', '127.0.0.1:%d' % port, binary],
stderr=subprocess.PIPE)
out = str(p.stderr.readline())
assert_equal(binary in out, True)
out = str(p.stderr.readline())
assert_equal(str(port) in out, True)
g = GDBProtocol(arch=avatar2.archs.X86_64)
g.remote_connect(port=port)
def wait_stopped(self):
# As we do not have access to avatar synchronizing target states
# on this level, we apply this little hack to synchronize the target
while True:
ret, out = self.gdb.console_command('info program')
if 'Program stopped' in out:
break
time.sleep(SLEEP_TIME)
def setup_inf_loop():
global p, g, port
binary = '%s/tests/binaries/infinite_loop' % os.getcwd()
p = subprocess.Popen(['gdbserver', '--once', '127.0.0.1:%d' % port, binary],
stderr=subprocess.PIPE)
out = str(p.stderr.readline())
assert_equal(binary in out, True)
out = str(p.stderr.readline())
assert_equal(str(port) in out, True)
g = GDBProtocol(arch=avatar2.archs.X86_64)
g.remote_connect(port=port)
def tearDown(self):
self.gdb.shutdown()
self.process.terminate()
def teardown_func():
g.shutdown()
p.terminate()
class GDBProtocolTestCaseOnHelloWorld(GdbProtocolTestCase):
@with_setup(setup_helloworld, teardown_func)
def test_register_names():
regs = g.get_register_names()
assert_list_equal(regs[:len(x86_regs)], x86_regs)
def setUp(self):
dir_path = dirname(realpath(__file__))
binary = '%s/binaries/hello_world' % dir_path
self.setup_env(binary)
@with_setup(setup_helloworld, teardown_func)
def test_register_read_and_write():
def test_register_names(self):
regs = self.gdb.get_register_names()
self.assertListEqual(regs[:len(X86_REGS)], X86_REGS)
ret = g.write_register('rax', 1678)
assert_equal(ret, True)
ret = g.read_register('rax')
assert_equal(ret, 1678)
def test_register_read_and_write(self):
ret = self.gdb.write_register('rax', 1678)
self.assertEqual(ret, True, ret)
ret = self.gdb.read_register('rax')
self.assertEqual(ret, 1678, ret)
def test_break_run_and_read_write_mem(self):
ret = self.gdb.set_breakpoint('main')
self.assertEqual(ret, True, ret)
ret = self.gdb.cont()
self.assertEqual(ret, True, ret)
self.wait_stopped()
ret = self.gdb.read_memory(self.base_address, 4)
self.assertEqual(ret, 0x464c457f, ret)
ret = self.gdb.write_memory(self.base_address, 4, 0x41414141)
self.assertEqual(ret, True, ret)
ret = self.gdb.read_memory(self.base_address, 4)
self.assertEqual(ret, 0x41414141, ret)
def test_watchpoint(self):
ret = self.gdb.set_watchpoint(self.base_address+0x754, read=True,
write=False)
self.assertEqual(ret, True, ret)
ret = self.gdb.cont()
self.assertEqual(ret, True, ret)
self.wait_stopped()
ret = self.gdb.read_memory(self.base_address, 4)
self.assertEqual(ret, 0x464c457f, ret)
@with_setup(setup_helloworld, teardown_func)
def test_break_run_and_read_write_mem():
class GDBProtocolWithUnixSocketTestCase(GDBProtocolTestCaseOnHelloWorld):
ret = g.set_breakpoint('main')
assert_equal(ret, True)
ret = g.cont()
assert_equal(ret, True)
# todo: enable waiting
time.sleep(SLEEP_TIME)
ret = g.read_memory(MEM_ADDR, 4)
assert_equal(ret, 0x464c457f)
ret = g.write_memory(MEM_ADDR, 4, 0x41414141)
assert_equal(ret, True)
ret = g.read_memory(MEM_ADDR, 4)
assert_equal(ret, 0x41414141)
@with_setup(setup_inf_loop, teardown_func)
def test_continue_stopping_stepping():
ret = g.cont()
assert_equal(ret, True)
def setUp(self):
dir_path = dirname(realpath(__file__))
binary = '%s/binaries/hello_world' % dir_path
self.setup_env(binary, unix_socket=True)
ret = g.stop()
assert_equal(ret, True)
class GDBProtocolTestCaseOnInfiniteLoop(GdbProtocolTestCase):
time.sleep(SLEEP_TIME)
def setUp(self):
dir_path = dirname(realpath(__file__))
binary = '%s/binaries/infinite_loop' % dir_path
self.setup_env(binary)
ret = g.step()
assert_equal(ret, True)
time.sleep(SLEEP_TIME)
def test_continue_stopping_stepping(self):
@with_setup(setup_helloworld, teardown_func)
def test_watchpoint():
ret = g.set_watchpoint(0x555555554754, read=True,
write=False)
assert_equal(ret, True)
ret = self.gdb.cont()
self.assertEqual(ret, True, ret)
ret = g.cont()
assert_equal(ret, True)
time.sleep(SLEEP_TIME)
ret = self.gdb.stop()
self.assertEqual(ret, True, ret)
self.wait_stopped()
ret = self.gdb.step()
self.assertEqual(ret, True, ret)
self.wait_stopped()
ret = g.read_memory(MEM_ADDR, 4)
assert_equal(ret, 0x464c457f)
if __name__ == '__main__':
setup_helloworld()
test_break_run_and_read_write_mem()
teardown_func()
unittest.main()
tests/test_inceptionprotocol.py
+38 -42
View File
@@ -1,14 +1,14 @@
from avatar2.protocols.inception import IPCortexM3
#import avatar2
import unittest
import struct
import ctypes
from nose.tools import *
from avatar2.protocols.inception import IPCortexM3
SLEEP_TIME = 1
MEM_ADDR = 0x12345678
i = None
@@ -110,49 +110,45 @@ class FakeIPCortexM3(IPCortexM3):
# ****************************************************************************
def setup():
global i
class InceptionProtocolTestCase(unittest.TestCase):
i = FakeIPCortexM3()
i.connect()
i.reset()
def setUp(self):
self.ip = FakeIPCortexM3()
self.ip.connect()
self.ip.reset()
def tearDown(self):
self.ip.shutdown()
def teardown():
i.shutdown()
def test_register_read_and_write(self):
ret = self.ip.write_register('R0', 2020)
self.assertEqual(ret, True, ret)
ret = self.ip.read_register('r0')
self.assertEqual(ret, 0xdeadbeef, ret)
def test_break_run_and_read_write_mem(self):
ret = self.ip.set_breakpoint(0x8000000)
self.assertEqual(ret, True, ret)
ret = self.ip.cont()
self.assertEqual(ret, True, ret)
#time.sleep(SLEEP_TIME)
ret = self.ip.read_memory(MEM_ADDR, 4)
self.assertEqual(ret, 0xdeadbeef, ret)
ret = self.ip.write_memory(MEM_ADDR, 4, 0x8badf00d)
self.assertEqual(ret, True, ret)
self.assertEqual(self.ip._fakecm3.fake_write_addr, MEM_ADDR, self.ip._fakecm3.fake_write_addr)
self.assertEqual(self.ip._fakecm3.fake_write_val, 0x8badf00d, self.ip._fakecm3.fake_write_val)
@with_setup(setup, teardown)
def test_register_read_and_write():
ret = i.write_register('R0', 2020)
assert_equal(ret, True)
ret = i.read_register('r0')
assert_equal(ret, 0xdeadbeef)
@with_setup(setup, teardown)
def test_break_run_and_read_write_mem():
ret = i.set_breakpoint(0x8000000)
assert_equal(ret, True)
ret = i.cont()
assert_equal(ret, True)
#time.sleep(SLEEP_TIME)
ret = i.read_memory(MEM_ADDR, 4)
assert_equal(ret, 0xdeadbeef)
ret = i.write_memory(MEM_ADDR, 4, 0x8badf00d)
assert_equal(ret, True)
assert_equal(i._fakecm3.fake_write_addr, MEM_ADDR)
assert_equal(i._fakecm3.fake_write_val, 0x8badf00d)
if __name__ == '__main__':
setup()
test_break_run_and_read_write_mem()
teardown()
unittest.main()
tests/test_pypandatarget.py
+133 -146
View File
@@ -1,27 +1,21 @@
import os
from os.path import abspath
from time import sleep
from nose.tools import *
import unittest
import socket
from os.path import dirname, realpath
from time import sleep
from avatar2 import *
from avatar2.peripherals.nucleo_usart import *
from pandare import ffi
PORT = 9997
TEST_STRING = bytearray(b'Hello World !\n')
panda = None
avatar = None
s = None
test_string = bytearray(b'Hello World !\n')
sram_dump = './tests/pyperipheral/sram_dump.bin'
rcc_dump = './tests/pyperipheral/rcc_dump.bin'
firmware = './tests/pyperipheral/firmware.bin'
dir_path = dirname(realpath(__file__))
sram_dump = '%s/pyperipheral/sram_dump.bin' % dir_path
rcc_dump = '%s/pyperipheral/rcc_dump.bin' % dir_path
firmware = '%s/pyperipheral/firmware.bin' % dir_path
with open(sram_dump, 'rb') as f:
sram_data = f.read()
@@ -30,156 +24,149 @@ with open(rcc_dump, 'rb') as f:
rcc_data = f.read()
@timed(7)
def setup_func():
# There is only one pypanda instance in the process space allowed.
# Unfortunately, we need a lot of hacks to make this working for CI tests.
# So, beware the globals!
global panda
global avatar
global s
if panda is None:
class PypandaTargetTestCase(unittest.TestCase):
panda = None
avatar = None
sk = None
@classmethod
def setUpClass(cls):
# Initiate the avatar-object
avatar = Avatar(output_directory='/tmp/avatar', arch=ARM_CORTEX_M3)
cls.avatar = Avatar(output_directory='/tmp/avatar', arch=ARM_CORTEX_M3)
panda = avatar.add_target(PyPandaTarget, gdb_port=1236,
entry_address=0x08005105)
cls.panda = cls.avatar.add_target(PyPandaTarget, gdb_port=1236,
entry_address=0x08005105)
# Define the various memory ranges and store references to them
avatar.add_memory_range(0x08000000, 0x1000000, file=firmware)
avatar.add_memory_range(0x20000000, 0x14000)
avatar.add_memory_range(0x40023000, 0x1000)
cls.avatar.add_memory_range(0x08000000, 0x1000000, file=firmware)
cls.avatar.add_memory_range(0x20000000, 0x14000)
cls.avatar.add_memory_range(0x40023000, 0x1000)
avatar.init_targets()
cls.avatar.init_targets()
panda.pypanda.register_pyperipheral(NucleoUSART('USART', 0x40004400, 0x100,
nucleo_usart_port=PORT))
cls.panda.pypanda.register_pyperipheral(NucleoUSART('USART', 0x40004400, 0x100,
nucleo_usart_port=PORT))
time.sleep(1)
cls.sk = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
cls.sk.connect(('127.0.0.1', PORT))
@classmethod
def tearDownClass(cls):
cls.panda.shutdown()
cls.avatar.shutdown()
cls.sk.close()
time.sleep(10)
#@timed(7)
def setUp(self):
self.recv_data = []
self.panda.write_memory(0x20000000, len(sram_data), sram_data, raw=True)
self.panda.write_memory(0x40023000, len(rcc_data), rcc_data, raw=True)
self.panda.regs.pc = 0x08005105
self.panda.regs.sp = 0x20014000
self.panda.bp(0x0800419c) # Termination breakpoint to avoid qemu dying
#@timed(8)
def tearDown(self):
pass
def test_nucleo_usart_read(self):
self.panda.cont()
data = self.sk.recv(len(TEST_STRING), socket.MSG_WAITALL)
self.assertEqual(data, TEST_STRING, data)
self.panda.wait()
def test_panda_callback(self):
recv_data = []
def cb(env, pc, addr, size, buf):
if addr == 0x40004404:
recv_data.append(buf[0])
cb_handle = self.panda.register_callback('mmio_before_write', cb)
self.panda.cont()
self.panda.wait()
self.assertEqual(bytearray(recv_data), TEST_STRING, bytearray(recv_data))
self.panda.disable_callback(cb_handle)
#@timed(11)
def test_panda_hook(self):
def hook(env, tb, hook):
#global recv_data
self.recv_data += [self.panda.pypanda.arch.get_reg(env, 'r0')]
self.panda.add_hook(0x8004622, hook) # return address of serial_putc
self.panda.cont()
self.panda.wait()
self.assertEqual(bytearray(self.recv_data), TEST_STRING, bytearray(self.recv_data))
self.panda.pypanda.plugins['hooks'].disable_hooking()
#@timed(10)
def test_nucleo_usart_debug_read(self):
# pyperipherals in debug mode need to be read from/written to directly
pyperiph = self.panda.pypanda.pyperipherals[0]
self.sk.send(b'Hello World')
reply = bytearray()
time.sleep(.1)
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect(('127.0.0.1', PORT))
while pyperiph.read_memory(0x40004400,4) & (1<<5) != 0:
reply.append(pyperiph.read_memory(0x40004404,4))
panda.write_memory(0x20000000, len(sram_data), sram_data, raw=True)
panda.write_memory(0x40023000, len(rcc_data), rcc_data, raw=True)
self.assertEqual(reply, b'Hello World', reply)
panda.regs.pc = 0x08005105
panda.regs.sp = 0x20014000
self.sk.send(b'Hello World')
reply = bytearray()
time.sleep(.1)
while pyperiph.read_memory(0x40004400,1) & (1<<5) != 0:
reply.append(pyperiph.read_memory(0x40004404,1))
panda.bp(0x0800419c) # Termination breakpoint to avoid qemu dying
@timed(8)
def teardown_func():
#panda.shutdown()
#avatar.shutdown()
#time.sleep(1)
pass
self.assertEqual(reply, b'Hello World', reply)
#@timed(11)
def test_nucleo_usart_debug_write(self):
pyperiph = self.panda.pypanda.pyperipherals[0]
@with_setup(setup_func, teardown_func)
def test_nucleo_usart_read():
panda.cont()
data = s.recv(len(test_string), socket.MSG_WAITALL)
assert_equal(data, test_string)
@with_setup(setup_func, teardown_func)
def test_panda_callback():
recv_data = []
def cb(env, pc, addr, size, buf):
if addr == 0x40004404:
recv_data.append(buf[0])
cb_handle = panda.register_callback('mmio_before_write', cb)
panda.cont()
panda.wait()
assert_equal(bytearray(recv_data), test_string)
panda.disable_callback(cb_handle)
recv_data = []
@timed(11)
@with_setup(setup_func, teardown_func)
def test_panda_hook():
def hook(env, tb, hook):
global recv_data
recv_data += [panda.pypanda.arch.get_reg(env, 'r0')]
panda.add_hook(0x8004622, hook) # return address of serial_putc
panda.cont()
panda.wait()
assert_equal(bytearray(recv_data), test_string)
panda.pypanda.plugins['hooks'].disable_hooking()
@timed(10)
@with_setup(setup_func, teardown_func)
def test_nucleo_usart_debug_read():
# pyperipherals in debug mode need to be read from/written to directly
pyperiph = panda.pypanda.pyperipherals[0]
s.send(b'Hello World')
reply = bytearray()
time.sleep(.1)
while pyperiph.read_memory(0x40004400,4) & (1<<5) != 0:
reply.append(pyperiph.read_memory(0x40004404,4))
assert_equal(reply, b'Hello World')
s.send(b'Hello World')
reply = bytearray()
time.sleep(.1)
while pyperiph.read_memory(0x40004400,1) & (1<<5) != 0:
reply.append(pyperiph.read_memory(0x40004404,1))
assert_equal(reply, b'Hello World')
@timed(11)
@with_setup(setup_func, teardown_func)
def test_nucleo_usart_debug_write():
pyperiph = panda.pypanda.pyperipherals[0]
time.sleep(1)
for c in test_string:
pyperiph.write_memory(0x40004404, 1, c)
reply = s.recv(len(test_string), socket.MSG_WAITALL)
assert_equal(reply, test_string)
time.sleep(.1)
for c in test_string:
pyperiph.write_memory(0x40004404, 4, c)
reply = s.recv(len(test_string), socket.MSG_WAITALL)
assert_equal(reply, test_string)
time.sleep(1)
for c in TEST_STRING:
pyperiph.write_memory(0x40004404, 1, c)
reply = self.sk.recv(len(TEST_STRING), socket.MSG_WAITALL)
self.assertEqual(reply, TEST_STRING, reply)
time.sleep(.1)
for c in TEST_STRING:
pyperiph.write_memory(0x40004404, 4, c)
reply = self.sk.recv(len(TEST_STRING), socket.MSG_WAITALL)
self.assertEqual(reply, TEST_STRING, reply)
if __name__ == '__main__':
#setup_func()
#test_panda_callback()
#teardown_func()
#setup_func()
#test_panda_hook()
#teardown_func()
pass
unittest.main()
tests/test_qemutarget.py
+128 -143
View File
@@ -1,3 +1,11 @@
import unittest
import tempfile
import os
import time
import intervaltree
import logging
from avatar2 import QemuTarget
from avatar2 import MemoryRange
from avatar2 import Avatar
@@ -6,23 +14,12 @@ from avatar2.targets import Target, TargetStates
from avatar2.message import *
import tempfile
import os
import time
import intervaltree
import logging
from nose.tools import *
TEST_DIR = '/tmp/testava'
ARM_BASE_ADDR = 0x08000000
MIPS_BASE_ADDR = 0x1fc00000
qemu = None
fake_target = None
ARCH = None
rom_addr = None
test_dir = '/tmp/testava'
arm_bin = (b'\x1e\x10\xa0\xe3' # mov r1, #0x1e
ARM_BIN = (b'\x1e\x10\xa0\xe3' # mov r1, #0x1e
b'\x00\x00\x20\xe0' # eor r0, r0, r0
b'\x01\x00\x80\xe2' # add r0, r0, #1
b'\x01\x00\x50\xe1' # cmp r0, r1
@@ -30,7 +27,7 @@ arm_bin = (b'\x1e\x10\xa0\xe3' # mov r1, #0x1e
b'\x00\x00\x00\xe0' # and r0, r0, r0
b'\x00\x00\x00\xe0') # and r0, r0, r0
mips_bin = (b'\x24\x09\x00\x1e' # addiu $t1, $zero, 0x1e
MIPS_BIN = (b'\x24\x09\x00\x1e' # addiu $t1, $zero, 0x1e
b'\x01\x08\x40\x26' # xor $t0, $t0, $t0
b'\x21\x08\x00\x01' # addi $t0, $t0, 1
b'\x15\x09\xff\xfe' # bne $t0, $t1, 8
@@ -38,7 +35,7 @@ mips_bin = (b'\x24\x09\x00\x1e' # addiu $t1, $zero, 0x1e
b'\x01\x08\x40\x24' # and $t0, $t0, $t0
b'\x01\x08\x40\x24') # and $t0, $t0, $t0
x86_bin = (b'\xbb\x1e\x00\x00\x00' # mov ebx, 0x1e
X86_BIN = (b'\xbb\x1e\x00\x00\x00' # mov ebx, 0x1e
b'\x31\xc0' # xor eax, eax
b'\x83\xc0\x01' # add eax, 1
b'\x39\xd8' # cmp eax, ebx
@@ -46,7 +43,7 @@ x86_bin = (b'\xbb\x1e\x00\x00\x00' # mov ebx, 0x1e
b'\x21\xc0' # and eax, eax
b'\x21\xc0') # and eax, eax
x86_64_bin = (b'\x48\xc7\xc3\x1e\x00\x00\x00' # mov rbx, 0x1e
X86_64_BIN = (b'\x48\xc7\xc3\x1e\x00\x00\x00' # mov rbx, 0x1e
b'\x48\x31\xc0' # xor rax, rax
b'\x48\x83\xc0\x01' # add rax, 1
b'\x48\x39\xd8' # cmp rax, rbx
@@ -73,144 +70,132 @@ class FakeTarget(object):
return True
def setup():
global qemu
global avatar
global fake_target
arch = setup_ARCH()
avatar = Avatar(arch=arch, output_directory=test_dir)
qemu = QemuTarget(avatar, name='qemu_test',
#firmware="./tests/binaries/qemu_arm_test",
firmware='%s/firmware' % test_dir,
)
fake_target = FakeTarget()
dev1 = avatar.add_memory_range(0x101f2000, 0x1000, 'dev1', forwarded=True,
forwarded_to=fake_target,
qemu_name='avatar-rmemory')
mem1 = avatar.add_memory_range(rom_addr, 0x1000, 'mem1',
#file='%s/tests/binaries/qemu_arm_test' %
# os.getcwd())
file='%s/firmware' % test_dir)
def setup_ARCH():
global ARCH
global rom_addr
ARCH = os.getenv('AVATAR2_ARCH')
if ARCH == 'ARM':
arch = ARM
firmware = arm_bin
rom_addr = 0x08000000
elif ARCH == 'MIPS':
arch = MIPS_24KF
firmware = mips_bin
rom_addr = 0x1fc00000
else:
assert False, 'Invalid Achitecture'
if not os.path.exists(test_dir): os.makedirs(test_dir)
with open('%s/firmware' % test_dir, 'wb') as f:
f.write(firmware)
return arch
def teardown():
global qemu
qemu.shutdown()
class QemuTargetTestCase(unittest.TestCase):
@with_setup(setup, teardown)
def test_initilization():
global qemu
def setUp(self):
self.setup_env()
qemu.init()
qemu.wait()
assert_equal(qemu.state, TargetStates.STOPPED)
def setup_env(self, gdb_unix_socket_path=None):
self.rom_addr = None
self.arch = None
self.setup_arch()
self.avatar = Avatar(arch=self.arch, output_directory=TEST_DIR, configure_logging=False)
self.qemu = QemuTarget(self.avatar, name='qemu_test',
#firmware="./tests/binaries/qemu_arm_test",
firmware='%s/firmware' % TEST_DIR,
gdb_unix_socket_path=gdb_unix_socket_path,
)
self.fake_target = FakeTarget()
dev1 = self.avatar.add_memory_range(0x101f2000, 0x1000, 'dev1', forwarded=True,
forwarded_to=self.fake_target,
qemu_name='avatar-rmemory')
mem1 = self.avatar.add_memory_range(self.rom_addr, 0x1000, 'mem1',
#file='%s/tests/binaries/qemu_arm_test' %
# os.getcwd())
file='%s/firmware' % TEST_DIR
)
def setup_arch(self):
ARCH = os.getenv('AVATAR2_ARCH')
if ARCH == 'ARM':
self.arch = ARM
self.rom_addr = ARM_BASE_ADDR
firmware = ARM_BIN
elif ARCH == 'MIPS':
self.arch = MIPS_24KF
self.rom_addr = MIPS_BASE_ADDR
firmware = MIPS_BIN
else:
self.assertTrue(False, 'Invalid Achitecture')
if not os.path.exists(TEST_DIR): os.makedirs(TEST_DIR)
with open('%s/firmware' % TEST_DIR, 'wb') as f:
f.write(firmware)
def tearDown(self):
self.qemu.shutdown()
@with_setup(setup, teardown)
def test_step():
global qemu
global ARCH
def test_initilization(self):
self.qemu.init()
self.qemu.wait()
self.assertEqual(self.qemu.state, TargetStates.STOPPED, self.qemu.state)
qemu.init()
qemu.wait()
def test_step(self):
self.qemu.init()
self.qemu.wait()
qemu.regs.pc=rom_addr
qemu.step()
assert_equal(qemu.regs.pc, rom_addr + 4)
self.qemu.regs.pc=self.rom_addr
self.qemu.step()
pc = self.qemu.regs.pc
self.assertEqual(pc, self.rom_addr + 4, pc)
def test_memory_read(self):
self.qemu.init()
self.qemu.wait()
mem = self.qemu.read_memory(self.rom_addr, 4)
if self.arch == ARM:
self.assertEqual(mem, 0xe3a0101e, mem)
elif self.arch == MIPS_24KF:
#self.assertEqual(mem, 0x2409001e, mem)
self.assertEqual(mem, 0x1e000924, mem)
else:
self.assertTrue(False, "Architecture not supported")
def test_memory_write(self):
self.qemu.init()
self.qemu.wait()
self.qemu.write_memory(self.rom_addr, 4, 0x41414141)
mem = self.qemu.read_memory(self.rom_addr, 4)
self.assertEqual(mem, 0x41414141, mem)
def test_remote_memory_write(self):
self.qemu.init()
self.qemu.wait()
remote_memory_write = self.qemu.write_memory(0x101f2000,4,0x41414141)
self.assertEqual(remote_memory_write, True)
addr = self.fake_target.fake_write_addr
size = self.fake_target.fake_write_size
val = self.fake_target.fake_write_val
self.assertEqual(addr, 0x101f2000, addr)
self.assertEqual(size, 4, size)
self.assertEqual(val, 0x41414141, val)
def test_remote_memory_read(self):
self.qemu.init()
self.qemu.wait()
self.assertEqual(self.qemu.state, TargetStates.STOPPED, self.qemu.state)
remote_memory_read = self.qemu.read_memory(0x101f2000,4)
self.assertEqual(remote_memory_read, 0xdeadbeef, remote_memory_read)
@with_setup(setup, teardown)
def test_memory_read():
global qemu
global ARCH
class QemuTargetWithUnixSocketTestCase(QemuTargetTestCase):
qemu.init()
qemu.wait()
def setUp(self):
self.setup_env(gdb_unix_socket_path="/tmp/test_sock")
mem = qemu.read_memory(rom_addr,4)
#def test_initialization():
# self.qemu.init()
# self.qemu.wait()
# self.assertEqual(qemu.state, TargetStates.STOPPED, self.qemu.state)
if ARCH == 'ARM':
assert_equal(mem, 0xe3a0101e)
elif ARCH == 'MIPS':
#assert_equal(mem, 0x2409001e)
assert_equal(mem, 0x1e000924)
else:
assert False, "Architecture not supported"
@with_setup(setup, teardown)
def test_memory_write():
global qemu
qemu.init()
qemu.wait()
qemu.write_memory(rom_addr,4, 0x41414141)
mem = qemu.read_memory(rom_addr,4)
assert_equal(mem, 0x41414141)
@with_setup(setup, teardown)
def test_remote_memory_write():
global qemu
global avatar
qemu.init()
qemu.wait()
remote_memory_write = qemu.write_memory(0x101f2000,4,0x41414141)
assert_equal(remote_memory_write, True)
assert_equal(fake_target.fake_write_addr, 0x101f2000)
assert_equal(fake_target.fake_write_size, 4)
assert_equal(fake_target.fake_write_val, 0x41414141)
@with_setup(setup, teardown)
def test_remote_memory_read():
global qemu
global avatar
qemu.init()
qemu.wait()
assert_equal(qemu.state, TargetStates.STOPPED)
remote_memory_read = qemu.read_memory(0x101f2000,4)
assert_equal(remote_memory_read, 0xdeadbeef)
if __name__ == '__main__':
setup()
#test_remote_memory()
test_initilization()
teardown()
unittest.main()
tests/test_remote_memoryprotocol.py
+37 -32
View File
@@ -1,3 +1,5 @@
import unittest
import sys
if sys.version_info < (3, 0):
import Queue as queue
@@ -9,45 +11,48 @@ import avatar2.protocols.remote_memory
from os import O_CREAT, O_RDONLY, O_WRONLY, O_RDWR
from posix_ipc import MessageQueue
from nose.tools import *
def setup_func():
pass
class RemoteMemoryTestCase(unittest.TestCase):
def setUp(self):
pass
def tearDown(self):
pass
def teardown_func():
pass
def test_remote_memory_request(self):
magic1 = 0x4141414141414141
magic2 = 0x4242424242424242
magic3 = 0x4343434343434343
magic4 = 0x4444444444444444
magic5 = 4
q = queue.Queue()
a = MessageQueue('/a', flags=O_CREAT|O_RDWR)
b = MessageQueue('/b', flags=O_CREAT|O_RDWR)
mprot1 = avatar2.protocols.remote_memory.RemoteMemoryProtocol('/a','/b', q)
ret = mprot1.connect()
self.assertEqual(ret, True)
self.assertNotEqual(mprot1._rx_queue, None)
self.assertNotEqual(mprot1._tx_queue, None)
@with_setup(setup_func, teardown_func)
def test_remote_memory_request():
magic1 = 0x4141414141414141
magic2 = 0x4242424242424242
magic3 = 0x4343434343434343
magic4 = 0x4444444444444444
magic5 = 4
request = avatar2.protocols.remote_memory.RemoteMemoryReq(magic1, magic2,
magic3, magic4,
magic5, 1)
a.send(request)
msg = q.get()
q = queue.Queue()
a = MessageQueue('/a', flags=O_CREAT|O_RDWR)
b = MessageQueue('/b', flags=O_CREAT|O_RDWR)
mprot1 = avatar2.protocols.remote_memory.RemoteMemoryProtocol('/a','/b', q)
ret = mprot1.connect()
assert_equal(ret, True)
assert_not_equal(mprot1._rx_queue, None)
assert_not_equal(mprot1._tx_queue, None)
self.assertEqual(msg.id, magic1)
self.assertEqual(msg.pc, magic2)
self.assertEqual(msg.address, magic3)
self.assertEqual(msg.value, magic4)
self.assertEqual(msg.size, magic5)
request = avatar2.protocols.remote_memory.RemoteMemoryReq(magic1, magic2,
magic3, magic4,
magic5, 1)
a.send(request)
msg = q.get()
assert_equal(msg.id, magic1)
assert_equal(msg.pc, magic2)
assert_equal(msg.address, magic3)
assert_equal(msg.value, magic4)
assert_equal(msg.size, magic5)
if __name__ == '__main__':
unittest.main()
tests/utils.py
+50
View File
@@ -0,0 +1,50 @@
import socket
import threading
import os
import time
# oneshot UNIX to TCP socket proxy. Stops after first connection
def unix2tcp(unix_socket_path, tcp_host, tcp_port):
try:
os.unlink(unix_socket_path)
except OSError:
pass
usock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
tsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
usock.bind(unix_socket_path)
usock.listen(1)
def proxy_loop():
uconn, addr = usock.accept()
tsock.connect((tcp_host, tcp_port))
uconn.setblocking(False)
tsock.setblocking(False)
while True:
data = None
try:
data = uconn.recv(1000)
if len(data) == 0:
break
tsock.sendall(data)
except BlockingIOError:
pass
try:
data = tsock.recv(1000)
if len(data) == 0:
break
uconn.sendall(data)
except BlockingIOError:
pass
usock.close()
uconn.close()
tsock.close()
threading.Thread(target=proxy_loop, daemon=True).start()