mirror of
https://github.com/avatartwo/avatar2.git
synced 2026-10-11 22:29:59 +00:00
Improve CI (#98)
* New action to publish on release the python package on Pypi * Migrate tests to unittest and pytest * Rework the CI pipeline to build and test on 3 python versions (3.7, 3.8 and 3.9) * Rename Dockerfile into avatar2.Dockerfile Co-authored-by: mariusmue <[email protected]>
This commit is contained in:
rawsample
@@ -1,70 +0,0 @@
|
||||
name: Build and test
|
||||
|
||||
on:
|
||||
push:
|
||||
branches:
|
||||
- 'main'
|
||||
- 'dev/*'
|
||||
pull_request:
|
||||
branches:
|
||||
- 'main'
|
||||
- 'dev/*'
|
||||
|
||||
jobs:
|
||||
build_and_test:
|
||||
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: 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'
|
||||
@@ -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 build/qemu
|
||||
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
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
[](https://badge.fury.io/py/avatar2)
|
||||
[](https://travis-ci.org/avatartwo/avatar2)
|
||||
[](https://github.com/avatartwo/avatar2/actions/workflows/continuous_integration.yml)
|
||||
|
||||
<img src="./avatar_logo.png" width="40%" height="40%"><br/>
|
||||
|
||||
|
||||
+16
-1
@@ -1,9 +1,14 @@
|
||||
#!/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"
|
||||
|
||||
if [[ "$ID" == "ubuntu" ]]
|
||||
then
|
||||
@@ -28,6 +33,14 @@ else
|
||||
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
|
||||
|
||||
@@ -36,7 +49,9 @@ git submodule update --init dtc
|
||||
|
||||
mkdir -p ../../build/qemu/
|
||||
cd ../../build/qemu
|
||||
../../src/avatar-qemu/configure --disable-sdl --target-list=arm-softmmu
|
||||
../../src/avatar-qemu/configure \
|
||||
--disable-sdl \
|
||||
--target-list=$TARGET_LIST
|
||||
|
||||
make -j $QEMU_NPROC
|
||||
|
||||
|
||||
+109
-103
@@ -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()
|
||||
|
||||
+107
-94
@@ -1,131 +1,144 @@
|
||||
import unittest
|
||||
|
||||
from avatar2.protocols.gdb import GDBProtocol
|
||||
import avatar2
|
||||
|
||||
import subprocess
|
||||
import os
|
||||
import time
|
||||
import re
|
||||
|
||||
from nose.tools import *
|
||||
from os.path import dirname, realpath
|
||||
|
||||
SLEEP_TIME = 1
|
||||
|
||||
MEM_ADDR = 0x555555554000
|
||||
port = 4444
|
||||
p = None
|
||||
g = None
|
||||
SLEEP_TIME = .1
|
||||
|
||||
x86_regs = [u'rax', u'rbx', u'rcx', u'rdx', u'rsi', u'rdi', u'rbp', u'rsp',
|
||||
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):
|
||||
|
||||
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)
|
||||
self.gdb.remote_connect(port=PORT)
|
||||
|
||||
# let's resolve the base address of the binary
|
||||
ret, out = self.gdb.console_command("p &main")
|
||||
main_addr = int(re.search("0x[0-9a-f]+", out).group(0), 16)
|
||||
self.base_address = main_addr - main_addr % 0x1000
|
||||
|
||||
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)
|
||||
|
||||
|
||||
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 tearDown(self):
|
||||
self.gdb.shutdown()
|
||||
self.process.terminate()
|
||||
|
||||
|
||||
def setup_inf_loop():
|
||||
global p, g, port
|
||||
class GDBProtocolTestCaseOnHelloWorld(GdbProtocolTestCase):
|
||||
|
||||
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 setUp(self):
|
||||
dir_path = dirname(realpath(__file__))
|
||||
binary = '%s/binaries/hello_world' % dir_path
|
||||
self.setup_env(binary)
|
||||
|
||||
|
||||
def teardown_func():
|
||||
g.shutdown()
|
||||
p.terminate()
|
||||
def test_register_names(self):
|
||||
regs = self.gdb.get_register_names()
|
||||
|
||||
self.assertListEqual(regs[:len(X86_REGS)], X86_REGS)
|
||||
|
||||
@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 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_register_read_and_write():
|
||||
|
||||
ret = g.write_register('rax', 1678)
|
||||
assert_equal(ret, True)
|
||||
ret = g.read_register('rax')
|
||||
assert_equal(ret, 1678)
|
||||
class GDBProtocolTestCaseOnInfiniteLoop(GdbProtocolTestCase):
|
||||
|
||||
|
||||
@with_setup(setup_helloworld, teardown_func)
|
||||
def test_break_run_and_read_write_mem():
|
||||
|
||||
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/infinite_loop' % dir_path
|
||||
self.setup_env(binary)
|
||||
|
||||
|
||||
ret = g.stop()
|
||||
assert_equal(ret, True)
|
||||
def test_continue_stopping_stepping(self):
|
||||
|
||||
time.sleep(SLEEP_TIME)
|
||||
ret = self.gdb.cont()
|
||||
self.assertEqual(ret, True, ret)
|
||||
|
||||
ret = g.step()
|
||||
assert_equal(ret, True)
|
||||
|
||||
time.sleep(SLEEP_TIME)
|
||||
ret = self.gdb.stop()
|
||||
self.assertEqual(ret, True, ret)
|
||||
|
||||
@with_setup(setup_helloworld, teardown_func)
|
||||
def test_watchpoint():
|
||||
ret = g.set_watchpoint(0x555555554754, read=True,
|
||||
write=False)
|
||||
assert_equal(ret, True)
|
||||
self.wait_stopped()
|
||||
|
||||
ret = g.cont()
|
||||
assert_equal(ret, True)
|
||||
ret = self.gdb.step()
|
||||
self.assertEqual(ret, True, ret)
|
||||
|
||||
self.wait_stopped()
|
||||
|
||||
|
||||
time.sleep(SLEEP_TIME)
|
||||
|
||||
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()
|
||||
|
||||
@@ -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()
|
||||
|
||||
+133
-146
@@ -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()
|
||||
|
||||
+107
-136
@@ -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,118 @@ 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):
|
||||
|
||||
qemu.init()
|
||||
qemu.wait()
|
||||
assert_equal(qemu.state, TargetStates.STOPPED)
|
||||
self.rom_addr = None
|
||||
self.arch = None
|
||||
self.setup_arch()
|
||||
|
||||
self.avatar = Avatar(arch=self.arch, output_directory=TEST_DIR)
|
||||
self.qemu = QemuTarget(self.avatar, name='qemu_test',
|
||||
#firmware="./tests/binaries/qemu_arm_test",
|
||||
firmware='%s/firmware' % TEST_DIR,
|
||||
)
|
||||
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)
|
||||
|
||||
@with_setup(setup, teardown)
|
||||
def test_memory_read():
|
||||
global qemu
|
||||
global ARCH
|
||||
def test_memory_read(self):
|
||||
self.qemu.init()
|
||||
self.qemu.wait()
|
||||
|
||||
qemu.init()
|
||||
qemu.wait()
|
||||
mem = self.qemu.read_memory(self.rom_addr, 4)
|
||||
|
||||
mem = qemu.read_memory(rom_addr,4)
|
||||
if self.arch == ARM:
|
||||
self.assertEqual(mem, 0xe3a0101e, mem)
|
||||
|
||||
if ARCH == 'ARM':
|
||||
assert_equal(mem, 0xe3a0101e)
|
||||
elif self.arch == MIPS_24KF:
|
||||
#self.assertEqual(mem, 0x2409001e, mem)
|
||||
self.assertEqual(mem, 0x1e000924, mem)
|
||||
else:
|
||||
self.assertTrue(False, "Architecture not supported")
|
||||
|
||||
elif ARCH == 'MIPS':
|
||||
#assert_equal(mem, 0x2409001e)
|
||||
assert_equal(mem, 0x1e000924)
|
||||
def test_memory_write(self):
|
||||
self.qemu.init()
|
||||
self.qemu.wait()
|
||||
|
||||
else:
|
||||
assert False, "Architecture not supported"
|
||||
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)
|
||||
|
||||
@with_setup(setup, teardown)
|
||||
def test_memory_write():
|
||||
global qemu
|
||||
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)
|
||||
|
||||
qemu.init()
|
||||
qemu.wait()
|
||||
def test_remote_memory_read(self):
|
||||
self.qemu.init()
|
||||
self.qemu.wait()
|
||||
self.assertEqual(self.qemu.state, TargetStates.STOPPED, self.qemu.state)
|
||||
|
||||
qemu.write_memory(rom_addr,4, 0x41414141)
|
||||
mem = qemu.read_memory(rom_addr,4)
|
||||
assert_equal(mem, 0x41414141)
|
||||
remote_memory_read = self.qemu.read_memory(0x101f2000,4)
|
||||
self.assertEqual(remote_memory_read, 0xdeadbeef, remote_memory_read)
|
||||
|
||||
|
||||
@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()
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user