Alex-Devera/tpotce
1
0
Fork 0
mirror of https://github.com/telekom-security/tpotce.git synced 2026-05-29 17:24:15 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

add tests for adbhoney, ciscoasa

Browse source
This commit is contained in:
t3chn0m4g3 2026-05-27 12:21:29 +02:00
parent d709cfd1fd
commit 83a42274ac
5 changed files with 1075 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

38
docker/_tests/README.md Normal file
View file

@ -0,0 +1,38 @@
# Docker Smoke Tests
This directory contains post-build smoke tests for T-Pot Docker images.
The tests expect images to exist locally. They do not build images, and they do
not touch production `data/` or `data_backup/` paths.
## Usage
```bash
./docker/_tests/run.sh --list
./docker/_tests/run.sh
./docker/_tests/run.sh adbhoney
./docker/_tests/run.sh ciscoasa
```
Common options:
```bash
./docker/_tests/run.sh --timeout 45 --bind-ip 127.0.0.1
./docker/_tests/run.sh --keep-artifacts adbhoney
```
Individual tests can also be run directly:
```bash
./docker/_tests/tests/adbhoney.sh
./docker/_tests/tests/adbhoney.sh --image dtagdevsec/adbhoney:24.04 --host-port 15555
./docker/_tests/tests/ciscoasa.sh --https-port 18443 --ike-port 15000
```
## Conventions
- Put one executable test script per honeypot in `tests/<service>.sh`.
- Source `lib/common.sh` for Docker, Compose, cleanup, and artifact helpers.
- Use temporary directories under `/tmp` for logs and downloads.
- Bind host ports to loopback by default and prefer dynamic host ports.
- Fail with a clear image build hint when the target image is missing.

256
docker/_tests/lib/common.sh Normal file
View file

@ -0,0 +1,256 @@
#!/usr/bin/env bash
set -Eeuo pipefail
TEST_LIB_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
TEST_ROOT="$(cd -- "${TEST_LIB_DIR}/.." && pwd)"
REPO_ROOT="$(cd -- "${TEST_ROOT}/../.." && pwd)"
DOCKER_ROOT="${REPO_ROOT}/docker"
TEST_TIMEOUT="${TEST_TIMEOUT:-30}"
TEST_BIND_IP="${TEST_BIND_IP:-127.0.0.1}"
TEST_KEEP_ARTIFACTS="${TEST_KEEP_ARTIFACTS:-false}"
TEST_TMP_ROOT=""
TEST_HARNESS_COMPOSE=""
TEST_PROJECT_NAME=""
TEST_CONTAINER_NAME=""
TEST_ARTIFACT_LOG_DIR=""
test_info() {
printf '==> %s\n' "$*"
}
test_ok() {
printf '[OK] %s\n' "$*"
}
test_die() {
printf '[FAIL] %s\n' "$*" >&2
exit 1
}
test_require_command() {
command -v "$1" >/dev/null 2>&1 || test_die "Required command not found: $1"
}
test_check_dependencies() {
[[ -n "${BASH_VERSION:-}" ]] || test_die "This test suite requires bash"
test_require_command docker
test_require_command python3
docker compose version >/dev/null 2>&1 || test_die "Docker Compose plugin is required"
docker info >/dev/null 2>&1 || test_die "Docker daemon is not accessible"
}
test_validate_timeout() {
[[ "${TEST_TIMEOUT}" =~ ^[0-9]+$ ]] || test_die "--timeout must be a number"
(( TEST_TIMEOUT >= 1 )) || test_die "--timeout must be at least 1"
}
test_validate_port() {
local port="$1"
[[ "${port}" =~ ^[0-9]+$ ]] || test_die "Port must be a number: ${port}"
(( port >= 1 && port <= 65535 )) || test_die "Port must be between 1 and 65535: ${port}"
}
test_read_compose_image() {
local service="$1"
local fallback="$2"
local compose_file="${DOCKER_ROOT}/${service}/docker-compose.yml"
local detected=""
if [[ -f "${compose_file}" ]]; then
detected="$(
sed -n 's/^[[:space:]]*image:[[:space:]]*//p' "${compose_file}" \
| head -n 1 \
| tr -d "\"'"
)"
fi
if [[ -n "${detected}" ]]; then
printf '%s\n' "${detected}"
else
printf '%s\n' "${fallback}"
fi
}
test_require_image() {
local image="$1"
local build_hint="$2"
docker image inspect "${image}" >/dev/null 2>&1 || test_die "Image not found: ${image}. Build it first, for example: ${build_hint}"
}
test_ensure_port_free() {
local bind_ip="$1"
local host_port="$2"
python3 - "${bind_ip}" "${host_port}" <<'PY'
import socket
import sys
host = sys.argv[1]
port = int(sys.argv[2])
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try:
sock.bind((host, port))
except OSError as exc:
print(f"{host}:{port} is not available: {exc}", file=sys.stderr)
sys.exit(1)
finally:
sock.close()
PY
}
test_ensure_udp_port_free() {
local bind_ip="$1"
local host_port="$2"
python3 - "${bind_ip}" "${host_port}" <<'PY'
import socket
import sys
host = sys.argv[1]
port = int(sys.argv[2])
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
sock.bind((host, port))
except OSError as exc:
print(f"{host}:{port}/udp is not available: {exc}", file=sys.stderr)
sys.exit(1)
finally:
sock.close()
PY
}
test_prepare_harness() {
local test_name="$1"
TEST_TMP_ROOT="$(mktemp -d "${TMPDIR:-/tmp}/tpot-${test_name}.XXXXXX")"
TEST_HARNESS_COMPOSE="${TEST_TMP_ROOT}/docker-compose.yml"
TEST_PROJECT_NAME="tpot-${test_name}-$(date +%s)-$$"
TEST_CONTAINER_NAME="${TEST_PROJECT_NAME}-service"
chmod 0777 "${TEST_TMP_ROOT}"
}
test_compose() {
docker compose -f "${TEST_HARNESS_COMPOSE}" -p "${TEST_PROJECT_NAME}" "$@"
}
test_wait_for_container() {
local deadline=$((SECONDS + TEST_TIMEOUT))
local state=""
while (( SECONDS < deadline )); do
state="$(docker inspect -f '{{.State.Status}}' "${TEST_CONTAINER_NAME}" 2>/dev/null || true)"
case "${state}" in
running)
return 0
;;
exited|dead)
return 1
;;
esac
sleep 1
done
return 1
}
test_get_mapped_port() {
local service="$1"
local container_port="$2"
local mapping=""
local deadline=$((SECONDS + TEST_TIMEOUT))
while (( SECONDS < deadline )); do
mapping="$(test_compose port "${service}" "${container_port}" 2>/dev/null | tail -n 1 || true)"
if [[ -z "${mapping}" && -n "${TEST_CONTAINER_NAME}" ]]; then
mapping="$(docker port "${TEST_CONTAINER_NAME}" "${container_port}" 2>/dev/null | tail -n 1 || true)"
fi
if [[ -n "${mapping}" ]]; then
printf '%s\n' "${mapping##*:}"
return 0
fi
sleep 1
done
return 1
}
test_wait_for_file_text() {
local text="$1"
local directory="$2"
local deadline=$((SECONDS + TEST_TIMEOUT))
while (( SECONDS < deadline )); do
if grep -R -F -- "${text}" "${directory}" >/dev/null 2>&1; then
return 0
fi
sleep 1
done
return 1
}
test_show_diagnostics() {
printf '\n[diagnostics] Container state\n' >&2
if [[ -n "${TEST_CONTAINER_NAME}" ]]; then
docker inspect -f 'status={{.State.Status}} exit={{.State.ExitCode}} error={{.State.Error}}' "${TEST_CONTAINER_NAME}" >&2 || true
else
printf 'No container name available.\n' >&2
fi
printf '\n[diagnostics] Docker logs\n' >&2
if [[ -n "${TEST_HARNESS_COMPOSE}" && -f "${TEST_HARNESS_COMPOSE}" ]]; then
test_compose logs --no-color --tail=120 >&2 || true
else
printf 'No temporary compose file available.\n' >&2
fi
if [[ -n "${TEST_ARTIFACT_LOG_DIR}" ]]; then
printf '\n[diagnostics] Test log artifacts\n' >&2
if [[ -d "${TEST_ARTIFACT_LOG_DIR}" ]]; then
find "${TEST_ARTIFACT_LOG_DIR}" -maxdepth 1 -type f -print | sort >&2 || true
while IFS= read -r file; do
printf '\n--- %s ---\n' "${file}" >&2
tail -n 80 "${file}" >&2 || true
done < <(find "${TEST_ARTIFACT_LOG_DIR}" -maxdepth 1 -type f -print | sort)
else
printf 'No temporary log directory available.\n' >&2
fi
fi
}
test_cleanup() {
local status=$?
trap - EXIT
if (( status != 0 )); then
test_show_diagnostics
fi
if [[ -n "${TEST_HARNESS_COMPOSE}" && -f "${TEST_HARNESS_COMPOSE}" ]]; then
test_compose down --volumes --remove-orphans >/dev/null 2>&1 || true
fi
if [[ "${TEST_KEEP_ARTIFACTS}" == "true" ]]; then
if [[ -n "${TEST_TMP_ROOT}" ]]; then
printf 'Artifacts kept at: %s\n' "${TEST_TMP_ROOT}" >&2
fi
elif [[ -n "${TEST_TMP_ROOT}" && -d "${TEST_TMP_ROOT}" ]]; then
rm -rf -- "${TEST_TMP_ROOT}"
fi
exit "${status}"
}
test_enable_cleanup() {
trap test_cleanup EXIT
trap 'exit 130' INT
}

148
docker/_tests/run.sh Executable file
View file

@ -0,0 +1,148 @@
#!/usr/bin/env bash
set -Eeuo pipefail
SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
TEST_DIR="${SCRIPT_DIR}/tests"
LIST_ONLY="false"
TEST_TIMEOUT="30"
TEST_BIND_IP="127.0.0.1"
TEST_KEEP_ARTIFACTS="false"
SELECTED_TESTS=()
usage() {
cat <<EOF
Usage: $0 [options] [test ...]
Run T-Pot Docker post-build smoke tests.
Options:
--list List available tests.
--timeout SEC Timeout passed to each test. Default: 30.
--bind-ip IP Host IP used by tests for loopback bindings. Default: 127.0.0.1.
--keep-artifacts Keep temporary compose files and logs for failed or passed tests.
-h, --help Show this help message.
Examples:
$0 --list
$0
$0 adbhoney
EOF
}
die() {
printf '[FAIL] %s\n' "$*" >&2
exit 1
}
parse_args() {
while [[ $# -gt 0 ]]; do
case "$1" in
--list)
LIST_ONLY="true"
shift
;;
--timeout)
[[ $# -ge 2 ]] || die "--timeout requires an argument"
TEST_TIMEOUT="$2"
shift 2
;;
--timeout=*)
TEST_TIMEOUT="${1#*=}"
shift
;;
--bind-ip)
[[ $# -ge 2 ]] || die "--bind-ip requires an argument"
TEST_BIND_IP="$2"
shift 2
;;
--bind-ip=*)
TEST_BIND_IP="${1#*=}"
shift
;;
--keep-artifacts)
TEST_KEEP_ARTIFACTS="true"
shift
;;
-h|--help)
usage
exit 0
;;
-*)
die "Unknown option: $1"
;;
*)
SELECTED_TESTS+=("$1")
shift
;;
esac
done
}
validate_args() {
[[ "${TEST_TIMEOUT}" =~ ^[0-9]+$ ]] || die "--timeout must be a number"
(( TEST_TIMEOUT >= 1 )) || die "--timeout must be at least 1"
}
list_tests() {
local test_file=""
find "${TEST_DIR}" -maxdepth 1 -type f -name '*.sh' -perm -u+x -print \
| sort \
| while IFS= read -r test_file; do
basename "${test_file}" .sh
done
}
test_path_for() {
local test_name="$1"
local test_path="${TEST_DIR}/${test_name}.sh"
[[ -x "${test_path}" ]] || return 1
printf '%s\n' "${test_path}"
}
main() {
parse_args "$@"
validate_args
if [[ "${LIST_ONLY}" == "true" ]]; then
list_tests
exit 0
fi
if [[ ${#SELECTED_TESTS[@]} -eq 0 ]]; then
mapfile -t SELECTED_TESTS < <(list_tests)
fi
[[ ${#SELECTED_TESTS[@]} -gt 0 ]] || die "No tests found in ${TEST_DIR}"
local common_args=(--timeout "${TEST_TIMEOUT}" --bind-ip "${TEST_BIND_IP}")
if [[ "${TEST_KEEP_ARTIFACTS}" == "true" ]]; then
common_args+=(--keep-artifacts)
fi
local passed=0
local failed=0
local test_name=""
local test_path=""
for test_name in "${SELECTED_TESTS[@]}"; do
test_path="$(test_path_for "${test_name}")" || die "Unknown test: ${test_name}"
printf '\n### Running %s\n' "${test_name}"
if "${test_path}" "${common_args[@]}"; then
printf '### PASS %s\n' "${test_name}"
passed=$((passed + 1))
else
printf '### FAIL %s\n' "${test_name}" >&2
failed=$((failed + 1))
fi
done
printf '\n### Summary: %s passed, %s failed\n' "${passed}" "${failed}"
(( failed == 0 ))
}
main "$@"

373
docker/_tests/tests/adbhoney.sh Executable file
View file

@ -0,0 +1,373 @@
#!/usr/bin/env bash
set -Eeuo pipefail
SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
# shellcheck source=../lib/common.sh
source "${SCRIPT_DIR}/../lib/common.sh"
TEST_NAME="adbhoney"
DEFAULT_IMAGE="dtagdevsec/adbhoney:24.04"
IMAGE=""
HOST_PORT=""
LOG_DIR=""
DL_DIR=""
TEXT_LOG_FILE=""
JSON_LOG_FILE=""
MAPPED_PORT=""
usage() {
cat <<EOF
Usage: $0 [options]
Run an isolated post-build smoke test for the ADBHoney image.
Options:
--image IMAGE Image to test. Defaults to docker/adbhoney/docker-compose.yml.
--host-port PORT Host port to bind. Default: dynamic free port.
--timeout SEC Timeout for startup, protocol, and log checks. Default: 30.
--bind-ip IP Host IP to bind. Default: 127.0.0.1.
--keep-artifacts Keep temporary compose file and logs for debugging.
-h, --help Show this help message.
EOF
}
parse_args() {
while [[ $# -gt 0 ]]; do
case "$1" in
--image)
[[ $# -ge 2 ]] || test_die "--image requires an argument"
IMAGE="$2"
shift 2
;;
--image=*)
IMAGE="${1#*=}"
shift
;;
--host-port|--port)
[[ $# -ge 2 ]] || test_die "$1 requires an argument"
HOST_PORT="$2"
shift 2
;;
--host-port=*|--port=*)
HOST_PORT="${1#*=}"
shift
;;
--timeout)
[[ $# -ge 2 ]] || test_die "--timeout requires an argument"
TEST_TIMEOUT="$2"
shift 2
;;
--timeout=*)
TEST_TIMEOUT="${1#*=}"
shift
;;
--bind-ip)
[[ $# -ge 2 ]] || test_die "--bind-ip requires an argument"
TEST_BIND_IP="$2"
shift 2
;;
--bind-ip=*)
TEST_BIND_IP="${1#*=}"
shift
;;
--keep-artifacts)
TEST_KEEP_ARTIFACTS="true"
shift
;;
-h|--help)
usage
exit 0
;;
*)
test_die "Unknown option: $1"
;;
esac
done
}
validate_args() {
test_validate_timeout
if [[ -n "${HOST_PORT}" ]]; then
test_validate_port "${HOST_PORT}"
fi
}
prepare_adbhoney_harness() {
test_prepare_harness "${TEST_NAME}"
LOG_DIR="${TEST_TMP_ROOT}/log"
DL_DIR="${TEST_TMP_ROOT}/downloads"
TEXT_LOG_FILE="${LOG_DIR}/adbhoney.log"
JSON_LOG_FILE="${LOG_DIR}/adbhoney.json"
TEST_ARTIFACT_LOG_DIR="${LOG_DIR}"
mkdir -p "${LOG_DIR}" "${DL_DIR}"
chmod 0777 "${LOG_DIR}" "${DL_DIR}"
cat > "${TEST_HARNESS_COMPOSE}" <<EOF
services:
adbhoney:
image: "${IMAGE}"
container_name: "${TEST_CONTAINER_NAME}"
restart: "no"
read_only: true
user: "2000:2000"
ports:
- "${TEST_BIND_IP}:${HOST_PORT}:5555"
volumes:
- "${LOG_DIR}:/opt/adbhoney/log"
- "${DL_DIR}:/opt/adbhoney/dl"
networks:
default:
name: "${TEST_PROJECT_NAME}_net"
EOF
}
wait_for_adbhoney_start_log() {
local deadline=$((SECONDS + TEST_TIMEOUT))
while (( SECONDS < deadline )); do
if [[ -f "${TEXT_LOG_FILE}" ]] && grep -F -- "Listening on 0.0.0.0:5555." "${TEXT_LOG_FILE}" >/dev/null 2>&1; then
return 0
fi
sleep 1
done
return 1
}
run_adb_probe() {
local token="$1"
python3 - "${TEST_BIND_IP}" "${MAPPED_PORT}" "${token}" "${TEST_TIMEOUT}" <<'PY'
import socket
import struct
import sys
import time
host = sys.argv[1]
port = int(sys.argv[2])
token = sys.argv[3]
timeout = int(sys.argv[4])
command_ids = {
name: struct.unpack("<I", name.encode("ascii"))[0]
for name in ("SYNC", "CNXN", "AUTH", "OPEN", "OKAY", "CLSE", "WRTE")
}
command_names = {value: name for name, value in command_ids.items()}
class ProbeError(Exception):
pass
def packet(command, arg0=0, arg1=0, payload=b""):
if isinstance(payload, str):
payload = payload.encode("utf-8")
command_id = command_ids[command]
checksum = sum(payload) & 0xFFFFFFFF
magic = command_id ^ 0xFFFFFFFF
header = struct.pack("<6I", command_id, arg0, arg1, len(payload), checksum, magic)
return header + payload
def recv_exact(sock, size, deadline):
chunks = []
remaining = size
while remaining > 0:
remaining_time = deadline - time.monotonic()
if remaining_time <= 0:
raise ProbeError(f"Timed out while waiting for {size} bytes")
sock.settimeout(min(remaining_time, 1.0))
try:
chunk = sock.recv(remaining)
except socket.timeout:
continue
if not chunk:
raise ProbeError("Connection closed by peer")
chunks.append(chunk)
remaining -= len(chunk)
return b"".join(chunks)
def recv_packet(sock, deadline):
header = recv_exact(sock, 24, deadline)
command_id, arg0, arg1, length, checksum, magic = struct.unpack("<6I", header)
if magic != (command_id ^ 0xFFFFFFFF):
raise ProbeError(f"Invalid ADB magic for command id {command_id:#x}")
if length > 1024 * 1024:
raise ProbeError(f"Refusing oversized ADB payload: {length} bytes")
payload = recv_exact(sock, length, deadline) if length else b""
actual_checksum = sum(payload) & 0xFFFFFFFF
if actual_checksum != checksum:
raise ProbeError(
f"Invalid ADB checksum for {command_names.get(command_id, hex(command_id))}: "
f"expected {checksum:#x}, got {actual_checksum:#x}"
)
return command_names.get(command_id, hex(command_id)), arg0, arg1, payload
deadline = time.monotonic() + timeout
client_id = 1
banner = None
remote_id = None
try:
with socket.create_connection((host, port), timeout=timeout) as sock:
sock.sendall(packet("CNXN", 0x01000000, 4096, b"host::adbhoney-test\0"))
while time.monotonic() < deadline:
command, arg0, arg1, payload = recv_packet(sock, deadline)
if command == "CNXN":
banner = payload.decode("utf-8", errors="replace").rstrip("\0")
break
if command == "AUTH":
raise ProbeError("ADBHoney requested AUTH; expected unauthenticated CNXN")
if not banner:
raise ProbeError("No CNXN response received")
destination = f"shell:echo {token}\0".encode("utf-8")
sock.sendall(packet("OPEN", client_id, 0, destination))
while time.monotonic() < deadline:
command, arg0, arg1, payload = recv_packet(sock, deadline)
if command == "OKAY" and arg1 == client_id:
remote_id = arg0
break
if command == "WRTE" and arg1 == client_id:
remote_id = arg0
break
if command == "CLSE" and arg1 == client_id:
raise ProbeError("ADBHoney closed the shell stream before accepting OPEN")
if remote_id is None:
raise ProbeError("No OKAY/WRTE response received for shell OPEN")
print(f"ADB CNXN banner: {banner}")
print(f"ADB shell OPEN accepted with remote id: {remote_id}")
except (OSError, ProbeError) as exc:
print(f"ADB probe failed: {exc}", file=sys.stderr)
sys.exit(1)
PY
}
run_adb_probe_with_retries() {
local token="$1"
local deadline=$((SECONDS + TEST_TIMEOUT))
local output=""
while (( SECONDS < deadline )); do
if output="$(run_adb_probe "${token}" 2>&1)"; then
printf '%s\n' "${output}"
return 0
fi
sleep 1
done
printf '%s\n' "${output}" >&2
return 1
}
wait_for_json_command_event() {
local token="$1"
python3 - "${JSON_LOG_FILE}" "${token}" "${TEST_TIMEOUT}" <<'PY'
import json
import sys
import time
from pathlib import Path
json_log = Path(sys.argv[1])
token = sys.argv[2]
timeout = int(sys.argv[3])
deadline = time.monotonic() + timeout
expected_input = f"echo {token}"
last_error = None
while time.monotonic() < deadline:
if json_log.exists():
try:
lines = json_log.read_text(encoding="utf-8").splitlines()
except OSError as exc:
last_error = f"Could not read {json_log}: {exc}"
time.sleep(1)
continue
for line_number, line in enumerate(lines, 1):
if not line.strip():
continue
try:
event = json.loads(line)
except json.JSONDecodeError as exc:
last_error = f"Invalid JSON in {json_log}:{line_number}: {exc}"
continue
if (
event.get("eventid") == "adbhoney.command.input"
and event.get("input") == expected_input
and token in event.get("input", "")
):
print(f"JSON event found: {event['eventid']} input={event['input']}")
sys.exit(0)
else:
last_error = f"{json_log} does not exist yet"
time.sleep(1)
if last_error:
print(last_error, file=sys.stderr)
print(f"No adbhoney.command.input event found in {json_log} for token {token}", file=sys.stderr)
sys.exit(1)
PY
}
main() {
parse_args "$@"
validate_args
test_check_dependencies
if [[ -z "${IMAGE}" ]]; then
IMAGE="$(test_read_compose_image "${TEST_NAME}" "${DEFAULT_IMAGE}")"
fi
test_info "Using image: ${IMAGE}"
test_require_image "${IMAGE}" "docker compose -f docker/${TEST_NAME}/docker-compose.yml build ${TEST_NAME}"
if [[ -n "${HOST_PORT}" ]]; then
test_ensure_port_free "${TEST_BIND_IP}" "${HOST_PORT}" || test_die "${TEST_BIND_IP}:${HOST_PORT} is already in use"
fi
prepare_adbhoney_harness
test_enable_cleanup
test_info "Starting isolated ADBHoney container"
test_compose up -d --no-build >/dev/null
test_wait_for_container || test_die "ADBHoney container did not stay running"
test_ok "Container is running"
test_info "Waiting for ADBHoney listener entry in adbhoney.log"
wait_for_adbhoney_start_log || test_die "ADBHoney listener entry was not found in adbhoney.log"
test_ok "ADBHoney listener entry found in adbhoney.log"
MAPPED_PORT="$(test_get_mapped_port "${TEST_NAME}" "5555")" || test_die "Could not resolve mapped host port for 5555/tcp"
test_ok "Port ${TEST_BIND_IP}:${MAPPED_PORT} maps to container port 5555/tcp"
local token="adbhoney-test-$(date +%s)-$$"
test_info "Running ADB protocol probe with token: ${token}"
run_adb_probe_with_retries "${token}" || test_die "ADB protocol probe failed on ${TEST_BIND_IP}:${MAPPED_PORT}"
test_ok "ADB protocol probe succeeded"
test_info "Waiting for command event in adbhoney.json"
wait_for_json_command_event "${token}" || test_die "Command event was not found in adbhoney.json"
test_ok "Command event was written to adbhoney.json"
test_ok "ADBHoney post-build smoke test completed successfully"
}
main "$@"

260
docker/_tests/tests/ciscoasa.sh Executable file
View file

@ -0,0 +1,260 @@
#!/usr/bin/env bash
set -Eeuo pipefail
SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
# shellcheck source=../lib/common.sh
source "${SCRIPT_DIR}/../lib/common.sh"
TEST_NAME="ciscoasa"
DEFAULT_IMAGE="dtagdevsec/ciscoasa:24.04"
IMAGE=""
HTTPS_PORT="8443"
IKE_PORT="5000"
LOG_DIR=""
LOG_FILE=""
MAPPED_HTTPS_PORT=""
MAPPED_IKE_PORT=""
usage() {
cat <<EOF
Usage: $0 [options]
Run an isolated post-build smoke test for the CiscoASA image.
Options:
--image IMAGE Image to test. Defaults to docker/ciscoasa/docker-compose.yml.
--https-port PORT Host TCP port for HTTPS. Default: 8443.
--ike-port PORT Host UDP port for IKE. Default: 5000.
--timeout SEC Timeout for startup, protocol, and log checks. Default: 30.
--bind-ip IP Host IP to bind. Default: 127.0.0.1.
--keep-artifacts Keep temporary compose file and logs for debugging.
-h, --help Show this help message.
EOF
}
parse_args() {
while [[ $# -gt 0 ]]; do
case "$1" in
--image)
[[ $# -ge 2 ]] || test_die "--image requires an argument"
IMAGE="$2"
shift 2
;;
--image=*)
IMAGE="${1#*=}"
shift
;;
--https-port)
[[ $# -ge 2 ]] || test_die "--https-port requires an argument"
HTTPS_PORT="$2"
shift 2
;;
--https-port=*)
HTTPS_PORT="${1#*=}"
shift
;;
--ike-port)
[[ $# -ge 2 ]] || test_die "--ike-port requires an argument"
IKE_PORT="$2"
shift 2
;;
--ike-port=*)
IKE_PORT="${1#*=}"
shift
;;
--timeout)
[[ $# -ge 2 ]] || test_die "--timeout requires an argument"
TEST_TIMEOUT="$2"
shift 2
;;
--timeout=*)
TEST_TIMEOUT="${1#*=}"
shift
;;
--bind-ip)
[[ $# -ge 2 ]] || test_die "--bind-ip requires an argument"
TEST_BIND_IP="$2"
shift 2
;;
--bind-ip=*)
TEST_BIND_IP="${1#*=}"
shift
;;
--keep-artifacts)
TEST_KEEP_ARTIFACTS="true"
shift
;;
-h|--help)
usage
exit 0
;;
*)
test_die "Unknown option: $1"
;;
esac
done
}
validate_args() {
test_validate_timeout
test_validate_port "${HTTPS_PORT}"
test_validate_port "${IKE_PORT}"
}
prepare_ciscoasa_harness() {
test_prepare_harness "${TEST_NAME}"
LOG_DIR="${TEST_TMP_ROOT}/log"
LOG_FILE="${LOG_DIR}/ciscoasa.log"
TEST_ARTIFACT_LOG_DIR="${LOG_DIR}"
mkdir -p "${LOG_DIR}"
chmod 0777 "${LOG_DIR}"
cat > "${TEST_HARNESS_COMPOSE}" <<EOF
services:
ciscoasa:
image: "${IMAGE}"
container_name: "${TEST_CONTAINER_NAME}"
restart: "no"
read_only: true
user: "2000:2000"
tmpfs:
- /tmp/ciscoasa:uid=2000,gid=2000
ports:
- "${TEST_BIND_IP}:${IKE_PORT}:5000/udp"
- "${TEST_BIND_IP}:${HTTPS_PORT}:8443"
volumes:
- "${LOG_DIR}:/var/log/ciscoasa"
networks:
default:
name: "${TEST_PROJECT_NAME}_net"
EOF
}
wait_for_log_line() {
local pattern="$1"
local deadline=$((SECONDS + TEST_TIMEOUT))
while (( SECONDS < deadline )); do
if [[ -f "${LOG_FILE}" ]] && grep -F -- "${pattern}" "${LOG_FILE}" >/dev/null 2>&1; then
return 0
fi
sleep 1
done
return 1
}
run_https_probe() {
local token="$1"
python3 - "${TEST_BIND_IP}" "${HTTPS_PORT}" "${token}" "${TEST_TIMEOUT}" <<'PY'
import socket
import ssl
import sys
import time
host = sys.argv[1]
port = int(sys.argv[2])
token = sys.argv[3]
timeout = int(sys.argv[4])
deadline = time.monotonic() + timeout
path = f"/+CSCOE+/logon.html?{token}"
request = (
f"GET {path} HTTP/1.1\r\n"
f"Host: {host}\r\n"
f"User-Agent: tpot-ciscoasa-smoke/{token}\r\n"
"Connection: close\r\n"
"\r\n"
).encode("ascii")
context = ssl._create_unverified_context()
try:
with socket.create_connection((host, port), timeout=timeout) as raw_sock:
with context.wrap_socket(raw_sock, server_hostname=host) as sock:
sock.settimeout(1)
sock.sendall(request)
chunks = []
while time.monotonic() < deadline:
try:
chunk = sock.recv(4096)
except socket.timeout:
continue
if not chunk:
break
chunks.append(chunk)
response = b"".join(chunks)
if not response.startswith(b"HTTP/"):
raise RuntimeError(f"Expected HTTP response, got {response[:80]!r}")
status_line = response.splitlines()[0].decode("iso-8859-1", errors="replace")
print(f"HTTPS response: {status_line}")
except Exception as exc:
print(f"HTTPS probe failed: {exc}", file=sys.stderr)
sys.exit(1)
PY
}
assert_no_runtime_errors() {
if [[ -f "${LOG_FILE}" ]] && grep -E "Traceback|NameError|Exception in callback" "${LOG_FILE}" >/dev/null 2>&1; then
test_die "CiscoASA runtime error found in ciscoasa.log"
fi
}
main() {
parse_args "$@"
validate_args
test_check_dependencies
if [[ -z "${IMAGE}" ]]; then
IMAGE="$(test_read_compose_image "${TEST_NAME}" "${DEFAULT_IMAGE}")"
fi
test_info "Using image: ${IMAGE}"
test_require_image "${IMAGE}" "docker compose -f docker/${TEST_NAME}/docker-compose.yml build ${TEST_NAME}"
test_ensure_port_free "${TEST_BIND_IP}" "${HTTPS_PORT}" || test_die "${TEST_BIND_IP}:${HTTPS_PORT} is already in use. Try --https-port <free-port>."
test_ensure_udp_port_free "${TEST_BIND_IP}" "${IKE_PORT}" || test_die "${TEST_BIND_IP}:${IKE_PORT}/udp is already in use. Try --ike-port <free-port>."
prepare_ciscoasa_harness
test_enable_cleanup
test_info "Starting isolated CiscoASA container"
test_compose up -d --no-build >/dev/null
test_wait_for_container || test_die "CiscoASA container did not stay running"
test_ok "Container is running"
test_info "Waiting for CiscoASA HTTPS listener in ciscoasa.log"
wait_for_log_line "Starting server on port 8443/tcp" || test_die "HTTPS listener entry was not found in ciscoasa.log"
test_ok "HTTPS listener entry found in ciscoasa.log"
test_info "Waiting for CiscoASA IKE listener in ciscoasa.log"
wait_for_log_line "Starting server on port 5000/udp" || test_die "IKE listener entry was not found in ciscoasa.log"
test_ok "IKE listener entry found in ciscoasa.log"
MAPPED_HTTPS_PORT="$(test_get_mapped_port "${TEST_NAME}" "8443")" || test_die "Could not resolve mapped host port for 8443/tcp"
test_ok "Port ${TEST_BIND_IP}:${MAPPED_HTTPS_PORT} maps to container port 8443/tcp"
MAPPED_IKE_PORT="$(test_get_mapped_port "${TEST_NAME}" "5000/udp")" || test_die "Could not resolve mapped host port for 5000/udp"
test_ok "Port ${TEST_BIND_IP}:${MAPPED_IKE_PORT} maps to container port 5000/udp"
local token="ciscoasa-test-$(date +%s)-$$"
test_info "Running HTTPS probe with token: ${token}"
run_https_probe "${token}"
test_wait_for_file_text "${token}" "${LOG_DIR}" || test_die "HTTPS probe token was not found in ciscoasa.log"
test_ok "HTTPS probe was written to ciscoasa.log"
test_wait_for_container || test_die "CiscoASA container stopped after HTTPS probe"
assert_no_runtime_errors
test_ok "No CiscoASA runtime errors found in ciscoasa.log"
test_ok "CiscoASA post-build smoke test completed successfully"
}
main "$@"