mirror of
https://github.com/trailofbits/buttercup
synced 2026-06-21 14:11:39 +00:00
Riccardo Schirone
885912d755
The subprocess mock was missing __enter__ and __exit__ configuration, causing tests to hang when ChallengeTask._run_cmd used `with subprocess.Popen(...) as process:` context manager syntax. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
1773 lines
67 KiB
Python
1773 lines
67 KiB
Python
import base64
|
|
import os
|
|
import platform
|
|
import subprocess
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from dirty_equals import IsStr
|
|
|
|
from buttercup.common.challenge_task import (
|
|
ChallengeTask,
|
|
ChallengeTaskError,
|
|
CommandResult,
|
|
ReproduceResult,
|
|
)
|
|
from buttercup.common.task_meta import TaskMeta
|
|
|
|
# ruff: noqa: E501, W291
|
|
|
|
|
|
@pytest.fixture
|
|
def task_dir(tmp_path: Path) -> Path:
|
|
"""Create a mock challenge task directory structure."""
|
|
# Create the main directories
|
|
tmp_path = tmp_path / "task-id-challenge-task"
|
|
oss_fuzz = tmp_path / "fuzz-tooling" / "my-oss-fuzz"
|
|
source = tmp_path / "src" / "my-source"
|
|
diffs = tmp_path / "diff" / "my-diff"
|
|
|
|
oss_fuzz.mkdir(parents=True, exist_ok=True)
|
|
source.mkdir(parents=True, exist_ok=True)
|
|
diffs.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Create some mock patch files
|
|
(diffs / "patch1.diff").write_text("mock patch 1")
|
|
(diffs / "patch2.diff").write_text("mock patch 2")
|
|
|
|
# Create a mock helper.py file
|
|
helper_path = oss_fuzz / "infra/helper.py"
|
|
helper_path.parent.mkdir(parents=True, exist_ok=True)
|
|
helper_path.write_text("import sys;\nsys.exit(0)\n")
|
|
|
|
# Create a mock test.txt file
|
|
(source / "test.txt").write_text("mock test content")
|
|
|
|
# Create task metadata
|
|
TaskMeta(
|
|
project_name="example_project",
|
|
focus="my-source",
|
|
task_id="task-id-challenge-task",
|
|
metadata={"task_id": "task-id-challenge-task", "round_id": "testing", "team_id": "tob"},
|
|
).save(tmp_path)
|
|
|
|
return tmp_path
|
|
|
|
|
|
@pytest.fixture
|
|
def challenge_task_readonly(task_dir: Path) -> ChallengeTask:
|
|
"""Create a mock challenge task for testing."""
|
|
return ChallengeTask(
|
|
read_only_task_dir=task_dir,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def challenge_task(task_dir: Path) -> ChallengeTask:
|
|
"""Create a mock challenge task for testing."""
|
|
return ChallengeTask(
|
|
read_only_task_dir=task_dir,
|
|
local_task_dir=task_dir,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def challenge_task_custom_python(task_dir: Path) -> ChallengeTask:
|
|
"""Create a mock challenge task with custom python path for testing."""
|
|
return ChallengeTask(
|
|
read_only_task_dir=task_dir,
|
|
local_task_dir=task_dir,
|
|
python_path="/usr/bin/python3",
|
|
)
|
|
|
|
|
|
def get_mock_popen(returncode: int, stdout: list[bytes], stderr: list[bytes]):
|
|
with patch("subprocess.Popen") as mock_popen:
|
|
mock_process = MagicMock()
|
|
mock_process.returncode = returncode
|
|
|
|
# Create mock pipe for stdout that handles multiple readline() calls gracefully
|
|
class MockStd:
|
|
def __init__(self, lines: list[bytes]):
|
|
self.lines = lines
|
|
self.current_line = 0
|
|
|
|
def readline(self):
|
|
if self.current_line < len(self.lines):
|
|
line = self.lines[self.current_line]
|
|
self.current_line += 1
|
|
return line
|
|
return b"" # Return empty bytes when no more lines
|
|
|
|
mock_process.stdout = MockStd(stdout)
|
|
|
|
# Create mock pipe for stderr
|
|
mock_process.stderr = MockStd(stderr)
|
|
|
|
# Setup poll and wait behavior
|
|
mock_process.poll.side_effect = [None, None, returncode]
|
|
mock_process.wait.return_value = returncode
|
|
|
|
# Make Popen return our mock process and work as context manager
|
|
mock_popen.return_value = mock_process
|
|
mock_popen.return_value.__enter__ = MagicMock(return_value=mock_process)
|
|
mock_popen.return_value.__exit__ = MagicMock(return_value=False)
|
|
yield mock_popen
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_subprocess():
|
|
"""Mock subprocess calls for testing."""
|
|
yield from get_mock_popen(0, [b"output line 1\n", b"output line 2\n"], [b"error line 1\n", b"error line 2\n"])
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_failed_subprocess():
|
|
"""Mock subprocess calls for testing."""
|
|
yield from get_mock_popen(1, [], [b"Build failed"])
|
|
|
|
|
|
def test_directory_structure(challenge_task: ChallengeTask):
|
|
"""Test that the challenge task correctly identifies its directory structure."""
|
|
assert challenge_task.get_oss_fuzz_subpath() == Path("fuzz-tooling") / "my-oss-fuzz"
|
|
assert challenge_task.get_source_subpath() == Path("src") / "my-source"
|
|
assert challenge_task.get_diff_subpath() == Path("diff") / "my-diff"
|
|
|
|
assert (challenge_task.task_dir / challenge_task.get_oss_fuzz_subpath()).is_dir()
|
|
assert (challenge_task.task_dir / challenge_task.get_source_subpath()).is_dir()
|
|
assert (challenge_task.task_dir / challenge_task.get_diff_subpath()).is_dir()
|
|
|
|
assert challenge_task.get_source_path() == challenge_task.task_dir / challenge_task.get_source_subpath()
|
|
assert challenge_task.get_diff_path() == challenge_task.task_dir / challenge_task.get_diff_subpath()
|
|
assert challenge_task.get_oss_fuzz_path() == challenge_task.task_dir / challenge_task.get_oss_fuzz_subpath()
|
|
|
|
|
|
def test_readonly_task(challenge_task_readonly: ChallengeTask):
|
|
"""Test that a readonly task raises an error when trying to build."""
|
|
with pytest.raises(ChallengeTaskError, match="Challenge Task is read-only, cannot perform this operation"):
|
|
challenge_task_readonly.build_image()
|
|
|
|
with pytest.raises(ChallengeTaskError, match="Challenge Task is read-only, cannot perform this operation"):
|
|
challenge_task_readonly.build_fuzzers(engine="libfuzzer", sanitizer="address")
|
|
|
|
with pytest.raises(ChallengeTaskError, match="Challenge Task is read-only, cannot perform this operation"):
|
|
challenge_task_readonly.reproduce_pov(fuzzer_name="fuzz_target", crash_path=Path("crash-sample"))
|
|
|
|
|
|
def test_build_image(challenge_task: ChallengeTask, mock_subprocess):
|
|
"""Test building the docker image for the project."""
|
|
result = challenge_task.build_image()
|
|
|
|
assert result.success is True
|
|
assert result.output is not None
|
|
assert result.output == b"output line 1\noutput line 2\n"
|
|
mock_subprocess.assert_called_once()
|
|
|
|
# Verify the command and working directory
|
|
args, kwargs = mock_subprocess.call_args
|
|
assert args[0] == [
|
|
"python",
|
|
"infra/helper.py",
|
|
"build_image",
|
|
"--no-pull",
|
|
"--architecture",
|
|
IsStr,
|
|
"example_project",
|
|
]
|
|
assert kwargs["cwd"] == challenge_task.task_dir / challenge_task.get_oss_fuzz_subpath()
|
|
|
|
# Verify output was read
|
|
mock_process = mock_subprocess.return_value
|
|
assert mock_process.wait.called # Verify we waited for process completion
|
|
|
|
|
|
def test_build_fuzzers(challenge_task: ChallengeTask, mock_subprocess):
|
|
"""Test building fuzzers with different configurations."""
|
|
result = challenge_task.build_fuzzers(
|
|
engine="libfuzzer",
|
|
sanitizer="address",
|
|
)
|
|
|
|
assert result.success is True
|
|
mock_subprocess.assert_called_once()
|
|
args, kwargs = mock_subprocess.call_args
|
|
assert args[0][:-1] == [
|
|
"python",
|
|
"infra/helper.py",
|
|
"build_fuzzers",
|
|
"--architecture",
|
|
IsStr,
|
|
"--engine",
|
|
"libfuzzer",
|
|
"--sanitizer",
|
|
"address",
|
|
"example_project",
|
|
]
|
|
assert args[0][-1].endswith(str(challenge_task.get_source_subpath()))
|
|
assert kwargs["cwd"] == challenge_task.task_dir / challenge_task.get_oss_fuzz_subpath()
|
|
|
|
|
|
def test_check_build(challenge_task: ChallengeTask, mock_subprocess):
|
|
"""Test checking the build status."""
|
|
result = challenge_task.check_build(engine="libfuzzer", sanitizer="address")
|
|
|
|
assert result.success is True
|
|
mock_subprocess.assert_called_once()
|
|
args, kwargs = mock_subprocess.call_args
|
|
assert args[0] == [
|
|
"python",
|
|
"infra/helper.py",
|
|
"check_build",
|
|
"--architecture",
|
|
IsStr,
|
|
"--engine",
|
|
"libfuzzer",
|
|
"--sanitizer",
|
|
"address",
|
|
"example_project",
|
|
]
|
|
assert kwargs["cwd"] == challenge_task.task_dir / challenge_task.get_oss_fuzz_subpath()
|
|
|
|
|
|
def test_reproduce_pov(challenge_task: ChallengeTask, mock_subprocess):
|
|
"""Test reproducing a proof of vulnerability."""
|
|
pov_file = Path("crash-sample")
|
|
result = challenge_task.reproduce_pov(
|
|
fuzzer_name="fuzz_target",
|
|
crash_path=pov_file,
|
|
)
|
|
|
|
assert result.did_crash() is False
|
|
mock_subprocess.assert_called_once()
|
|
args, kwargs = mock_subprocess.call_args
|
|
assert args[0][:-1] == [
|
|
"python",
|
|
"infra/helper.py",
|
|
"reproduce",
|
|
"--architecture",
|
|
IsStr,
|
|
"example_project",
|
|
"fuzz_target",
|
|
]
|
|
assert args[0][-1].endswith("/crash-sample")
|
|
assert kwargs["cwd"] == challenge_task.task_dir / challenge_task.get_oss_fuzz_subpath()
|
|
|
|
|
|
def test_failed_build_image(challenge_task: ChallengeTask, mock_failed_subprocess):
|
|
"""Test handling failed build image operation."""
|
|
result = challenge_task.build_image()
|
|
|
|
assert result.success is False
|
|
assert result.error == b"Build failed"
|
|
|
|
|
|
def test_invalid_task_dir():
|
|
"""Test handling of invalid task directory."""
|
|
with pytest.raises(ChallengeTaskError, match="Missing required directory: /nonexistent"):
|
|
ChallengeTask(
|
|
read_only_task_dir=Path("/nonexistent"),
|
|
)
|
|
|
|
|
|
def test_missing_required_dirs(tmp_path: Path):
|
|
"""Test handling of missing required directories."""
|
|
# Create task dir without required subdirectories
|
|
task_dir = tmp_path / "task"
|
|
task_dir.mkdir()
|
|
|
|
# Add TaskMeta even though directories are missing
|
|
TaskMeta(
|
|
project_name="example_project",
|
|
focus="my-source",
|
|
task_id="task-id-challenge-task",
|
|
metadata={"task_id": "task-id-challenge-task", "round_id": "testing", "team_id": "tob"},
|
|
).save(task_dir)
|
|
|
|
with pytest.raises(ChallengeTaskError, match=f"Missing required directory: {task_dir / 'src'}"):
|
|
ChallengeTask(
|
|
read_only_task_dir=task_dir,
|
|
)
|
|
|
|
|
|
def test_build_image_custom_python(challenge_task_custom_python: ChallengeTask, mock_subprocess):
|
|
"""Test building the docker image using custom python path."""
|
|
result = challenge_task_custom_python.build_image()
|
|
|
|
assert result.success is True
|
|
mock_subprocess.assert_called_once()
|
|
args, kwargs = mock_subprocess.call_args
|
|
assert args[0] == [
|
|
"/usr/bin/python3",
|
|
"infra/helper.py",
|
|
"build_image",
|
|
"--no-pull",
|
|
"--architecture",
|
|
IsStr,
|
|
"example_project",
|
|
]
|
|
assert kwargs["cwd"] == challenge_task_custom_python.task_dir / challenge_task_custom_python.get_oss_fuzz_subpath()
|
|
|
|
|
|
def test_build_fuzzers_custom_python(challenge_task_custom_python: ChallengeTask, mock_subprocess):
|
|
"""Test building fuzzers with custom python path."""
|
|
result = challenge_task_custom_python.build_fuzzers(
|
|
engine="libfuzzer",
|
|
sanitizer="address",
|
|
)
|
|
|
|
assert result.success is True
|
|
mock_subprocess.assert_called_once()
|
|
args, kwargs = mock_subprocess.call_args
|
|
assert args[0][:-1] == [
|
|
"/usr/bin/python3",
|
|
"infra/helper.py",
|
|
"build_fuzzers",
|
|
"--architecture",
|
|
IsStr,
|
|
"--engine",
|
|
"libfuzzer",
|
|
"--sanitizer",
|
|
"address",
|
|
"example_project",
|
|
]
|
|
assert args[0][-1].endswith(str(challenge_task_custom_python.get_source_subpath()))
|
|
assert kwargs["cwd"] == challenge_task_custom_python.task_dir / challenge_task_custom_python.get_oss_fuzz_subpath()
|
|
|
|
|
|
@pytest.fixture
|
|
def libjpeg_oss_fuzz_task_dir(tmp_path: Path) -> Path:
|
|
"""Create a challenge task using a real OSS-Fuzz repository."""
|
|
# Clone real oss-fuzz repo into temp dir
|
|
tmp_path = tmp_path / "libjpeg-turbo"
|
|
tmp_path.mkdir(parents=True)
|
|
|
|
oss_fuzz_dir = tmp_path / "fuzz-tooling"
|
|
oss_fuzz_dir.mkdir(parents=True)
|
|
source_dir = tmp_path / "src"
|
|
source_dir.mkdir(parents=True)
|
|
|
|
subprocess.run(["git", "-C", str(oss_fuzz_dir), "clone", "https://github.com/google/oss-fuzz.git"], check=True)
|
|
# Restore libjpeg-turbo project directory to specific commit
|
|
subprocess.run(
|
|
[
|
|
"git",
|
|
"-C",
|
|
str(oss_fuzz_dir / "oss-fuzz"),
|
|
"checkout",
|
|
"a5d517924f758028f299d7c6cecf3b471503a202",
|
|
"--",
|
|
"projects/libjpeg-turbo",
|
|
],
|
|
check=True,
|
|
)
|
|
|
|
# Download libpng source code
|
|
libjpeg_url = "https://github.com/libjpeg-turbo/libjpeg-turbo"
|
|
# Checkout specific libjpeg commit for reproducibility
|
|
subprocess.run(["git", "-C", str(source_dir), "clone", libjpeg_url], check=True)
|
|
subprocess.run(
|
|
["git", "-C", str(source_dir / "libjpeg-turbo"), "checkout", "6d91e950c871103a11bac2f10c63bf998796c719"],
|
|
check=True,
|
|
)
|
|
|
|
# Create task metadata
|
|
TaskMeta(
|
|
project_name="libjpeg-turbo",
|
|
focus="libjpeg-turbo",
|
|
task_id="task-id-libjpeg-turbo",
|
|
metadata={"task_id": "task-id-libjpeg-turbo", "round_id": "testing", "team_id": "tob"},
|
|
).save(tmp_path)
|
|
|
|
return tmp_path
|
|
|
|
|
|
@pytest.fixture
|
|
def libjpeg_oss_fuzz_task(libjpeg_oss_fuzz_task_dir: Path) -> ChallengeTask:
|
|
return ChallengeTask(
|
|
read_only_task_dir=libjpeg_oss_fuzz_task_dir,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def libjpeg_oss_fuzz_task_rw(libjpeg_oss_fuzz_task: ChallengeTask) -> ChallengeTask:
|
|
with libjpeg_oss_fuzz_task.get_rw_copy(None) as local_task:
|
|
yield local_task
|
|
|
|
|
|
@pytest.fixture
|
|
def libjpeg_crash_testcase() -> Path:
|
|
"""Get libjpeg-turbo crash testcase"""
|
|
crash_file = Path(__file__).parent / "data" / "libjpeg-crash"
|
|
if not crash_file.exists():
|
|
raise FileNotFoundError(f"Crash file not found at {crash_file}")
|
|
|
|
return crash_file
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_real_build_workflow(libjpeg_oss_fuzz_task_rw: ChallengeTask):
|
|
"""Test the full build workflow using actual OSS-Fuzz repository."""
|
|
# Build the base image
|
|
result = libjpeg_oss_fuzz_task_rw.build_image(pull_latest_base_image=False)
|
|
assert result.success is True, f"Build image failed: {result.error}"
|
|
|
|
# Build the fuzzers
|
|
result = libjpeg_oss_fuzz_task_rw.build_fuzzers(
|
|
engine="libfuzzer",
|
|
sanitizer="address",
|
|
architecture="x86_64",
|
|
)
|
|
assert result.success is True, f"Build fuzzers failed: {result.error}"
|
|
|
|
# Check the build
|
|
result = libjpeg_oss_fuzz_task_rw.check_build(
|
|
engine="libfuzzer",
|
|
sanitizer="address",
|
|
architecture="x86_64",
|
|
)
|
|
assert result.success is True, f"Check build failed: {result.error}"
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_build_fuzzers_with_cache(libjpeg_oss_fuzz_task_rw: ChallengeTask):
|
|
"""Test building fuzzers with cache."""
|
|
result = libjpeg_oss_fuzz_task_rw.build_fuzzers_with_cache(
|
|
engine="libfuzzer",
|
|
sanitizer="address",
|
|
)
|
|
assert result.success is True
|
|
assert result.output is not None
|
|
assert "cp /src/libjpeg_turbo_fuzzer_seed_corpus.zip /out/" in result.output.decode()
|
|
|
|
result = libjpeg_oss_fuzz_task_rw.build_fuzzers_with_cache(
|
|
engine="libfuzzer",
|
|
sanitizer="address",
|
|
)
|
|
assert result.success is True
|
|
assert result.output is not None
|
|
assert "Check build passed" in result.output.decode()
|
|
assert "cp /src/libjpeg_turbo_fuzzer_seed_corpus.zip /out/" not in result.output.decode()
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_real_reproduce_pov(libjpeg_oss_fuzz_task_rw: ChallengeTask, libjpeg_crash_testcase: Path):
|
|
"""Test the reproduce POV workflow using actual OSS-Fuzz repository."""
|
|
# Reproduce the POV
|
|
libjpeg_oss_fuzz_task_rw.build_image(pull_latest_base_image=False)
|
|
libjpeg_oss_fuzz_task_rw.build_fuzzers(engine="libfuzzer", sanitizer="address")
|
|
result = libjpeg_oss_fuzz_task_rw.reproduce_pov(
|
|
fuzzer_name="libjpeg_turbo_fuzzer",
|
|
crash_path=libjpeg_crash_testcase,
|
|
)
|
|
assert result.did_crash(), "Reproduce POV failed"
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.parametrize(
|
|
"oss_fuzz_commit,description",
|
|
[
|
|
("eb47f56bd15626ab4ae8727bc435148dd2f9857c^", "before_bug"),
|
|
("eb47f56bd15626ab4ae8727bc435148dd2f9857c", "bug_introduced"),
|
|
("HEAD", "current_head"),
|
|
],
|
|
)
|
|
def test_helper_py_patching_across_commits(
|
|
tmp_path: Path, libjpeg_crash_testcase: Path, oss_fuzz_commit: str, description: str
|
|
):
|
|
"""Test helper.py patching across different OSS-Fuzz commits.
|
|
|
|
This test validates that helper.py patching works correctly across different
|
|
OSS-Fuzz commits:
|
|
- Before the bug: eb47f56^ (should work without needing reproduce_impl patch)
|
|
- Bug introduced: eb47f56bd15626ab4ae8727bc435148dd2f9857c (needs patching)
|
|
- Current HEAD: Latest OSS-Fuzz (needs patching)
|
|
|
|
The patching should ensure reproduce_pov runs successfully on all commits.
|
|
"""
|
|
# Detect current architecture
|
|
current_arch = platform.machine()
|
|
# Normalize architecture names
|
|
if current_arch in ("arm64", "aarch64"):
|
|
current_arch = "aarch64"
|
|
elif current_arch in ("x86_64", "amd64"):
|
|
current_arch = "x86_64"
|
|
|
|
# Only test ARM64 on ARM64 platforms, x86_64 on x86_64 platforms
|
|
# For this test we use the current architecture
|
|
|
|
# Create task directory structure
|
|
task_dir = tmp_path / f"libjpeg-turbo-{description}"
|
|
task_dir.mkdir(parents=True)
|
|
|
|
oss_fuzz_dir = task_dir / "fuzz-tooling"
|
|
oss_fuzz_dir.mkdir(parents=True)
|
|
source_dir = task_dir / "src"
|
|
source_dir.mkdir(parents=True)
|
|
|
|
# Clone OSS-Fuzz at specific commit
|
|
subprocess.run(["git", "-C", str(oss_fuzz_dir), "clone", "https://github.com/google/oss-fuzz.git"], check=True)
|
|
subprocess.run(
|
|
["git", "-C", str(oss_fuzz_dir / "oss-fuzz"), "checkout", oss_fuzz_commit],
|
|
check=True,
|
|
)
|
|
|
|
# Clone libjpeg-turbo source (same commit as existing test)
|
|
libjpeg_url = "https://github.com/libjpeg-turbo/libjpeg-turbo"
|
|
subprocess.run(["git", "-C", str(source_dir), "clone", libjpeg_url], check=True)
|
|
subprocess.run(
|
|
["git", "-C", str(source_dir / "libjpeg-turbo"), "checkout", "6d91e950c871103a11bac2f10c63bf998796c719"],
|
|
check=True,
|
|
)
|
|
|
|
# Create task metadata
|
|
TaskMeta(
|
|
project_name="libjpeg-turbo",
|
|
focus="libjpeg-turbo",
|
|
task_id=f"task-id-libjpeg-turbo-{description}",
|
|
metadata={"task_id": f"task-id-libjpeg-turbo-{description}", "round_id": "testing", "team_id": "tob"},
|
|
).save(task_dir)
|
|
|
|
# Create challenge task
|
|
task = ChallengeTask(
|
|
read_only_task_dir=task_dir,
|
|
)
|
|
|
|
# Get RW copy and test
|
|
with task.get_rw_copy(None) as local_task:
|
|
# Build image and fuzzers
|
|
result = local_task.build_image(pull_latest_base_image=False, architecture=current_arch)
|
|
assert result.success is True, f"Build image failed: {result.error}"
|
|
|
|
result = local_task.build_fuzzers(engine="libfuzzer", sanitizer="address", architecture=current_arch)
|
|
assert result.success is True, f"Build fuzzers failed: {result.error}"
|
|
|
|
# Reproduce the POV - this should work with patching (the fuzzer should run successfully)
|
|
# Note: We're testing that reproduce_pov runs without errors, not necessarily that
|
|
# this specific crash test case triggers a crash (compatibility may vary by version)
|
|
result = local_task.reproduce_pov(
|
|
fuzzer_name="libjpeg_turbo_fuzzer",
|
|
crash_path=libjpeg_crash_testcase,
|
|
)
|
|
|
|
# The key is that reproduce_pov should RUN successfully (did_run returns True)
|
|
# This confirms the patching fixed the reproduce_impl issue
|
|
assert result.did_run(), (
|
|
f"Reproduce POV did not run for commit {oss_fuzz_commit} ({description}). "
|
|
f"Patching may not have worked correctly. Output: {result.command_result.output[:500] if result.command_result.output else 'None'}"
|
|
)
|
|
|
|
# Verify that helper.py was patched correctly
|
|
helper_path = local_task.get_oss_fuzz_path() / "infra" / "helper.py"
|
|
assert helper_path.exists(), "helper.py not found"
|
|
|
|
helper_content = helper_path.read_text()
|
|
|
|
# Check for the common fix: architecture=architecture instead of err_result
|
|
# The fix should change: return run_function(run_args, err_result)
|
|
# To: return run_function(run_args, architecture=architecture)
|
|
assert "architecture=architecture" in helper_content, (
|
|
f"helper.py patching failed: 'architecture=architecture' not found in helper.py for commit {oss_fuzz_commit}"
|
|
)
|
|
|
|
# On ARM64, verify ARM64-specific patches were applied
|
|
if current_arch == "aarch64":
|
|
# Check for ARM64 manifest tags
|
|
assert ":manifest-arm64v8" in helper_content, (
|
|
f"helper.py patching failed: ARM64 manifest tags not found for commit {oss_fuzz_commit}"
|
|
)
|
|
|
|
|
|
def test_copy_task(challenge_task_readonly: ChallengeTask, mock_subprocess):
|
|
"""Test copying a challenge task to a temporary directory."""
|
|
with patch.object(ChallengeTask, "_check_python_path"):
|
|
with challenge_task_readonly.get_rw_copy(None) as local_task:
|
|
# Verify the task directory is different
|
|
assert local_task.task_dir != challenge_task_readonly.task_dir
|
|
|
|
# Verify the file structure was copied
|
|
assert local_task.get_source_path().is_dir()
|
|
assert local_task.get_oss_fuzz_path().is_dir()
|
|
assert local_task.get_diff_path().is_dir()
|
|
|
|
# Verify file contents were copied
|
|
copied_file = local_task.get_source_path() / "test.txt"
|
|
assert copied_file.exists()
|
|
assert copied_file.read_text() == "mock test content"
|
|
|
|
# Verify we can modify the copy without affecting the original
|
|
new_content = "modified content"
|
|
copied_file.write_text(new_content)
|
|
assert copied_file.read_text() != (challenge_task_readonly.get_source_path() / "test.txt").read_text()
|
|
|
|
result = local_task.build_image()
|
|
assert result.success is True
|
|
assert result.output is not None
|
|
assert result.output == b"output line 1\noutput line 2\n"
|
|
mock_subprocess.assert_called_once()
|
|
|
|
# Verify the temporary directory was cleaned up
|
|
assert not local_task.task_dir.exists()
|
|
|
|
|
|
def test_commit_task(challenge_task_readonly: ChallengeTask, mock_subprocess):
|
|
"""Test committing a challenge task."""
|
|
with patch.object(ChallengeTask, "_check_python_path"):
|
|
with challenge_task_readonly.get_rw_copy(None) as local_task:
|
|
local_task.build_image()
|
|
local_task.build_fuzzers(engine="libfuzzer", sanitizer="address")
|
|
old_local_dir = local_task.task_dir
|
|
local_task.commit()
|
|
commited_local_dir = local_task.task_dir
|
|
assert commited_local_dir != old_local_dir
|
|
assert commited_local_dir.exists()
|
|
assert commited_local_dir.is_dir()
|
|
assert not old_local_dir.exists()
|
|
|
|
commited_task = ChallengeTask(
|
|
read_only_task_dir=commited_local_dir,
|
|
local_task_dir=commited_local_dir,
|
|
)
|
|
assert commited_task.get_source_path().exists()
|
|
assert commited_task.get_oss_fuzz_path().exists()
|
|
assert commited_task.get_diff_path().exists()
|
|
|
|
with pytest.raises(ChallengeTaskError, match="Missing required directory"):
|
|
ChallengeTask(
|
|
read_only_task_dir=old_local_dir,
|
|
local_task_dir=old_local_dir,
|
|
)
|
|
|
|
|
|
def test_commit_task_with_suffix(challenge_task_readonly: ChallengeTask, mock_subprocess, tmp_path: Path):
|
|
"""Test committing a challenge task with a suffix."""
|
|
with patch.object(ChallengeTask, "_check_python_path"):
|
|
with challenge_task_readonly.get_rw_copy(work_dir=tmp_path) as local_task:
|
|
local_task.commit(suffix="test")
|
|
|
|
with challenge_task_readonly.get_rw_copy(work_dir=tmp_path) as local_task:
|
|
with pytest.raises(ChallengeTaskError, match="Failed to commit task"):
|
|
local_task.commit(suffix="test")
|
|
|
|
|
|
def test_no_commit_task(challenge_task_readonly: ChallengeTask, mock_subprocess):
|
|
"""Test that a not-commited task cannot be accessed after the context manager is closed."""
|
|
with patch.object(ChallengeTask, "_check_python_path"):
|
|
with challenge_task_readonly.get_rw_copy(None) as local_task:
|
|
local_task.build_image()
|
|
local_task.build_fuzzers(engine="libfuzzer", sanitizer="address")
|
|
old_local_dir = local_task.task_dir
|
|
|
|
with pytest.raises(ChallengeTaskError, match="Missing required directory"):
|
|
ChallengeTask(
|
|
read_only_task_dir=old_local_dir,
|
|
local_task_dir=old_local_dir,
|
|
)
|
|
|
|
|
|
def test_get_diffs(challenge_task: ChallengeTask):
|
|
"""Test getting diffs from the source code."""
|
|
diffs = challenge_task.get_diffs()
|
|
assert len(diffs) == 2
|
|
assert diffs[0].name == "patch1.diff"
|
|
assert diffs[1].name == "patch2.diff"
|
|
|
|
|
|
def test_is_delta_mode(challenge_task: ChallengeTask):
|
|
"""Test checking if the task is in diff mode."""
|
|
assert challenge_task.is_delta_mode() is True
|
|
|
|
|
|
def test_apply_patch_diff(challenge_task: ChallengeTask):
|
|
"""Test applying a patch diff to the source code."""
|
|
diff_path = challenge_task.get_diff_path() / "patch1.diff"
|
|
challenge_task.get_diffs = MagicMock(return_value=[diff_path])
|
|
with diff_path.open("w") as f:
|
|
f.write(r"""diff -ru a/test.txt b/test.txt
|
|
--- a/test.txt 2025-02-18 14:27:44.815130716 +0000
|
|
+++ b/test.txt 2025-02-18 14:28:12.061424543 +0000
|
|
@@ -1 +1 @@
|
|
-mock test content
|
|
\ No newline at end of file
|
|
+modified content
|
|
\ No newline at end of file
|
|
""")
|
|
|
|
challenge_task.apply_patch_diff()
|
|
assert challenge_task.get_source_path().joinpath("test.txt").exists()
|
|
assert challenge_task.get_source_path().joinpath("test.txt").read_text() == "modified content"
|
|
|
|
|
|
def test_restore_task(challenge_task_readonly: ChallengeTask):
|
|
"""Test restoring a challenge task."""
|
|
with patch.object(ChallengeTask, "_check_python_path"):
|
|
with challenge_task_readonly.get_rw_copy(None) as local_task:
|
|
local_task.get_source_path().joinpath("a.txt").write_text("a")
|
|
local_task.get_source_path().joinpath("b.txt").write_text("b")
|
|
assert local_task.get_source_path().joinpath("a.txt").exists()
|
|
|
|
local_task.restore()
|
|
assert not local_task.get_source_path().joinpath("a.txt").exists()
|
|
assert not local_task.get_source_path().joinpath("b.txt").exists()
|
|
assert local_task.get_source_path().joinpath("test.txt").exists()
|
|
|
|
|
|
def test_restore_task_same_dir(challenge_task: ChallengeTask):
|
|
"""Test restoring a challenge task to the same directory."""
|
|
with patch.object(ChallengeTask, "_check_python_path"):
|
|
with challenge_task.get_rw_copy(None) as local_task:
|
|
assert local_task.task_dir != local_task.read_only_task_dir, "get_rw_copy should always create a copy"
|
|
local_task.get_source_path().joinpath("b.txt").write_text("b")
|
|
|
|
local_task.restore()
|
|
|
|
|
|
def test_workdir_from_dockerfile(challenge_task_readonly: ChallengeTask):
|
|
"""Test getting the workdir from the dockerfile."""
|
|
assert challenge_task_readonly.workdir_from_dockerfile() == Path("/src/example_project")
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_workdir_from_dockerfile_libjpeg(libjpeg_oss_fuzz_task: ChallengeTask):
|
|
"""Test getting the workdir from the dockerfile."""
|
|
assert libjpeg_oss_fuzz_task.workdir_from_dockerfile() == Path("/src/libjpeg-turbo")
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_exec_docker_cmd(libjpeg_oss_fuzz_task_rw: ChallengeTask):
|
|
"""Test executing a command inside the docker container."""
|
|
result = libjpeg_oss_fuzz_task_rw.exec_docker_cmd(["ls", "/"])
|
|
assert result.success is True
|
|
assert result.output is not None
|
|
assert "root" in result.output.decode()
|
|
|
|
result = libjpeg_oss_fuzz_task_rw.exec_docker_cmd(["ls", "/src"])
|
|
assert result.success is True
|
|
assert result.output is not None
|
|
assert "libjpeg-turbo" in result.output.decode()
|
|
|
|
# Test mounting directories
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
tmp_path = Path(tmp_dir)
|
|
|
|
# Create some test files
|
|
(tmp_path / "test1.txt").write_text("test1")
|
|
(tmp_path / "test2.txt").write_text("test2")
|
|
|
|
# Mount the temp dir to /mnt in container
|
|
mount_dirs = {tmp_path: Path("/mnt")}
|
|
|
|
result = libjpeg_oss_fuzz_task_rw.exec_docker_cmd(["ls", "/mnt"], mount_dirs=mount_dirs)
|
|
assert result.success is True
|
|
assert result.output is not None
|
|
assert "test1.txt" in result.output.decode()
|
|
assert "test2.txt" in result.output.decode()
|
|
|
|
# Verify we can read the file contents
|
|
result = libjpeg_oss_fuzz_task_rw.exec_docker_cmd(["cat", "/mnt/test1.txt"], mount_dirs=mount_dirs)
|
|
assert result.success is True
|
|
assert result.output is not None
|
|
assert result.output.decode() == "test1"
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_build_and_restore(libjpeg_oss_fuzz_task_rw: ChallengeTask):
|
|
"""Test building and restoring a challenge task."""
|
|
libjpeg_oss_fuzz_task_rw.build_image()
|
|
libjpeg_oss_fuzz_task_rw.build_fuzzers()
|
|
libjpeg_oss_fuzz_task_rw.commit()
|
|
|
|
# Modify the source code
|
|
libjpeg_oss_fuzz_task_rw.get_source_path().joinpath("test.txt").write_text("modified content")
|
|
|
|
libjpeg_oss_fuzz_task_rw.restore()
|
|
|
|
assert not libjpeg_oss_fuzz_task_rw.get_source_path().joinpath("test.txt").exists()
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_tmp_dir(libjpeg_oss_fuzz_task: ChallengeTask):
|
|
"""Test building fuzzers with cache."""
|
|
local_dir = None
|
|
with libjpeg_oss_fuzz_task.get_rw_copy(None) as local_task:
|
|
local_dir = local_task.task_dir
|
|
assert local_dir.exists()
|
|
assert local_dir.is_dir()
|
|
local_task.build_fuzzers()
|
|
|
|
assert not local_dir.exists()
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_container_image(libjpeg_oss_fuzz_task: ChallengeTask):
|
|
"""Test getting the container image."""
|
|
assert libjpeg_oss_fuzz_task.container_image() == "gcr.io/oss-fuzz/libjpeg-turbo"
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_container_image_custom_org(libjpeg_oss_fuzz_task_dir: Path):
|
|
"""Test getting the container image with a custom organization."""
|
|
with patch.dict(os.environ, {"OSS_FUZZ_CONTAINER_ORG": "myorg"}):
|
|
task = ChallengeTask(
|
|
read_only_task_dir=libjpeg_oss_fuzz_task_dir,
|
|
)
|
|
assert task.container_image() == "myorg/libjpeg-turbo"
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def mock_node_data_dir():
|
|
"""Set the NODE_DATA_DIR environment variable for all tests."""
|
|
with patch.dict(os.environ, {"NODE_DATA_DIR": "/test/node/data/dir"}):
|
|
yield
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def mock_node_local(monkeypatch, tmp_path: Path):
|
|
"""Mock the node_local module functions used by ChallengeTask."""
|
|
# Create a patch for buttercup.common.node_local's _get_root_path to return a valid path
|
|
with patch("buttercup.common.node_local._get_root_path", return_value=Path("/test/node/data/dir")):
|
|
# Create a patch for remote_archive_to_dir that just returns the path
|
|
with patch("buttercup.common.node_local.remote_archive_to_dir") as mock_remote_archive:
|
|
# Create a patch for scratch_path to return a valid path
|
|
with patch("buttercup.common.node_local.scratch_path", return_value=tmp_path / "node-local-scratch"):
|
|
# The remote_archive_to_dir function should just return the input path
|
|
mock_remote_archive.side_effect = lambda p: p
|
|
yield
|
|
|
|
|
|
@patch("buttercup.common.node_local._get_root_path")
|
|
@patch("buttercup.common.node_local.remote_archive_to_dir")
|
|
def test_challenge_task_with_node_local_storage_existing(
|
|
mock_remote_archive_to_dir,
|
|
mock_get_root_path,
|
|
mock_node_local_storage,
|
|
task_dir,
|
|
):
|
|
"""Test ChallengeTask behavior when using node_local and path exists."""
|
|
mock_get_root_path.return_value = Path(mock_node_local_storage)
|
|
# When path already exists, remote_archive_to_dir should not be called
|
|
mock_remote_archive_to_dir.side_effect = Exception("Should not be called")
|
|
|
|
# Create a path that exists in the node local storage
|
|
existing_path = Path(mock_node_local_storage) / "existing-task"
|
|
|
|
# The original Path.exists method to use for paths we don't specifically handle
|
|
original_exists = Path.exists
|
|
|
|
# Patch Path.exists to return specific values based on what's being checked
|
|
def mock_exists(self):
|
|
# For the main task directory and helper.py, return True
|
|
if str(self) == str(existing_path) or str(self).endswith("helper.py"):
|
|
return True
|
|
# For checking required directories
|
|
if str(self).endswith("src") or str(self).endswith("fuzz-tooling"):
|
|
return True
|
|
# For test.txt
|
|
if str(self).endswith("test.txt"):
|
|
return True
|
|
# For all other cases, call the original method
|
|
return original_exists(self)
|
|
|
|
# Mock is_dir to always return True for directory checks
|
|
def mock_is_dir(self):
|
|
return True
|
|
|
|
with patch.object(Path, "exists", mock_exists):
|
|
with patch.object(Path, "is_dir", mock_is_dir):
|
|
# Create a challenge task with this path - it should use the local version
|
|
with patch.object(ChallengeTask, "_check_python_path"):
|
|
with patch.object(TaskMeta, "load"):
|
|
task = ChallengeTask(read_only_task_dir=existing_path)
|
|
|
|
# Verify it's using the local path
|
|
assert task.read_only_task_dir == existing_path
|
|
|
|
# Verify mock_remote_archive_to_dir wasn't called
|
|
mock_remote_archive_to_dir.assert_not_called()
|
|
|
|
# Verify content is from the local copy
|
|
with patch.object(Path, "read_text", return_value="node local storage content"):
|
|
source_file = task.get_source_path() / "test.txt"
|
|
assert source_file.read_text() == "node local storage content"
|
|
|
|
|
|
@patch("buttercup.common.node_local._get_root_path")
|
|
@patch("buttercup.common.node_local.remote_archive_to_dir")
|
|
def test_challenge_task_with_node_local_storage_download(
|
|
mock_remote_archive_to_dir,
|
|
mock_get_root_path,
|
|
mock_node_local_storage,
|
|
task_dir,
|
|
):
|
|
"""Test ChallengeTask behavior when using node_local and path doesn't exist."""
|
|
mock_get_root_path.return_value = Path(mock_node_local_storage)
|
|
|
|
# Create a path that doesn't exist in the node local storage
|
|
non_existing_path = Path(mock_node_local_storage) / "non-existing-task"
|
|
downloaded_path = Path(mock_node_local_storage) / "downloaded-task"
|
|
|
|
# Create the directory structure for the downloaded path
|
|
downloaded_path.mkdir(exist_ok=True)
|
|
oss_fuzz = downloaded_path / "fuzz-tooling" / "my-oss-fuzz"
|
|
source = downloaded_path / "src" / "my-source"
|
|
oss_fuzz.mkdir(parents=True, exist_ok=True)
|
|
source.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Create a mock helper.py file
|
|
helper_path = oss_fuzz / "infra/helper.py"
|
|
helper_path.parent.mkdir(parents=True, exist_ok=True)
|
|
helper_path.write_text("import sys;\nsys.exit(0)\n")
|
|
|
|
# Create task metadata
|
|
TaskMeta(
|
|
project_name="example_project",
|
|
focus="my-source",
|
|
task_id="task-id-challenge-task",
|
|
metadata={"task_id": "task-id-challenge-task", "round_id": "testing", "team_id": "tob"},
|
|
).save(downloaded_path)
|
|
|
|
# Setup mock to return a downloaded path
|
|
mock_remote_archive_to_dir.return_value = downloaded_path
|
|
|
|
# Patch Path.exists to return False for the original path
|
|
# but True for the downloaded path and its contents
|
|
original_exists = Path.exists
|
|
|
|
def mock_exists(self):
|
|
# Non-existing path returns False
|
|
if str(self) == str(non_existing_path):
|
|
return False
|
|
# For helper.py file, return True
|
|
if str(self).endswith("helper.py"):
|
|
return True
|
|
# For checking directories
|
|
if str(self).endswith("src") or str(self).endswith("fuzz-tooling"):
|
|
return True
|
|
# For all other cases, call the original method
|
|
return original_exists(self)
|
|
|
|
with patch.object(Path, "exists", mock_exists):
|
|
# Create a challenge task with this path - it should trigger download
|
|
with patch.object(ChallengeTask, "_check_python_path"):
|
|
with patch.object(ChallengeTask, "_check_dir_exists"):
|
|
task = ChallengeTask(read_only_task_dir=non_existing_path)
|
|
|
|
# Verify remote_archive_to_dir was called with correct path
|
|
mock_remote_archive_to_dir.assert_called_once_with(non_existing_path)
|
|
|
|
# Verify it's using the downloaded path
|
|
assert task.read_only_task_dir == downloaded_path
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_node_local_storage(tmp_path: Path):
|
|
"""Setup a mock NODE_DATA_DIR environment for testing node local storage."""
|
|
node_data_dir = tmp_path / "node_data_dir"
|
|
node_data_dir.mkdir(exist_ok=True)
|
|
scratch_dir = node_data_dir / "scratch"
|
|
scratch_dir.mkdir(exist_ok=True)
|
|
|
|
# Create a pre-existing local path to simulate already downloaded data
|
|
local_task_path = node_data_dir / "existing-task"
|
|
local_task_path.mkdir(exist_ok=True)
|
|
|
|
# Copy task structure to local task path
|
|
oss_fuzz = local_task_path / "fuzz-tooling" / "my-oss-fuzz"
|
|
source = local_task_path / "src" / "my-source"
|
|
oss_fuzz.mkdir(parents=True, exist_ok=True)
|
|
source.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Create a mock helper.py file
|
|
helper_path = oss_fuzz / "infra/helper.py"
|
|
helper_path.parent.mkdir(parents=True, exist_ok=True)
|
|
helper_path.write_text("import sys;\nsys.exit(0)\n")
|
|
|
|
# Create a mock test.txt file with different content to verify which one is used
|
|
(source / "test.txt").write_text("node local storage content")
|
|
|
|
# Create task metadata
|
|
TaskMeta(
|
|
project_name="example_project",
|
|
focus="my-source",
|
|
task_id="task-id-challenge-task",
|
|
metadata={"task_id": "task-id-challenge-task", "round_id": "testing", "team_id": "tob"},
|
|
).save(local_task_path)
|
|
|
|
return node_data_dir
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_run_fuzzer_libjpeg(libjpeg_oss_fuzz_task_rw: ChallengeTask, tmp_path: Path):
|
|
"""Test running a fuzzer on a challenge task for 10 seconds."""
|
|
corpus_dir = tmp_path / "test-corpus"
|
|
corpus_dir.mkdir(exist_ok=True)
|
|
|
|
libjpeg_oss_fuzz_task_rw.build_fuzzers()
|
|
result = libjpeg_oss_fuzz_task_rw.run_fuzzer(
|
|
harness_name="libjpeg_turbo_fuzzer",
|
|
corpus_dir=corpus_dir,
|
|
fuzzer_args=["\\-max_total_time=10"],
|
|
)
|
|
# Can't assert success because it will be False if the fuzzer finds a crash
|
|
# assert result.success is True
|
|
assert result.output is not None
|
|
|
|
|
|
def test_reproduce_result_stacktrace():
|
|
"""Test the stacktrace method of ReproduceResult with various output sizes."""
|
|
# Test case 1: Short string (under 1MB limit)
|
|
short_output = b"INFO: Seed: 12345\nRunning normally\n==ERROR: AddressSanitizer: heap-buffer-overflow"
|
|
result1 = ReproduceResult(
|
|
command_result=CommandResult(success=False, returncode=1, output=short_output, error=None),
|
|
)
|
|
stacktrace1 = result1.stacktrace()
|
|
assert stacktrace1 is not None
|
|
assert stacktrace1 == short_output.decode("utf-8", errors="ignore")
|
|
assert len(stacktrace1) == len(short_output.decode("utf-8", errors="ignore"))
|
|
|
|
# Test case 2: Large string (over 1MB limit) - should be truncated
|
|
# Create a large output that exceeds the 1MB limit
|
|
MAX_OUTPUT_LEN = 1 * 1024 * 1024 # 1 MB
|
|
large_content = (
|
|
b"INFO: Seed: 12345\n" + b"A" * (MAX_OUTPUT_LEN + 1000) + b"\n==ERROR: AddressSanitizer: heap-buffer-overflow"
|
|
)
|
|
|
|
result2 = ReproduceResult(
|
|
command_result=CommandResult(success=False, returncode=1, output=large_content, error=None),
|
|
)
|
|
stacktrace2 = result2.stacktrace()
|
|
assert stacktrace2 is not None
|
|
|
|
# Verify the output is truncated to approximately MAX_OUTPUT_LEN
|
|
assert len(stacktrace2.encode("utf-8")) <= MAX_OUTPUT_LEN + 100 # Allow some flexibility for truncation marker
|
|
|
|
# Verify it contains the truncation marker
|
|
assert "...truncated" in stacktrace2
|
|
|
|
# Verify it contains both the beginning and end of the original content
|
|
assert "INFO: Seed: 12345" in stacktrace2
|
|
assert "==ERROR: AddressSanitizer: heap-buffer-overflow" in stacktrace2
|
|
|
|
# Verify the structure: start + truncation marker + end
|
|
parts = stacktrace2.split("...truncated")
|
|
assert len(parts) == 2
|
|
assert parts[0].startswith("INFO: Seed: 12345")
|
|
assert parts[1].endswith("==ERROR: AddressSanitizer: heap-buffer-overflow")
|
|
|
|
# Test case 3: Empty output
|
|
result3 = ReproduceResult(command_result=CommandResult(success=True, returncode=0, output=None, error=None))
|
|
stacktrace3 = result3.stacktrace()
|
|
assert stacktrace3 is None
|
|
|
|
# Test case 4: Exactly at the limit (1MB)
|
|
prefix = b"INFO: Seed: 12345\n"
|
|
suffix = b"\n==ERROR: AddressSanitizer: heap-buffer-overflow"
|
|
exact_size_content = prefix + b"B" * (MAX_OUTPUT_LEN - len(prefix) - len(suffix)) + suffix
|
|
|
|
result4 = ReproduceResult(
|
|
command_result=CommandResult(success=False, returncode=1, output=exact_size_content, error=None),
|
|
)
|
|
stacktrace4 = result4.stacktrace()
|
|
assert stacktrace4 is not None
|
|
|
|
# Should not be truncated since it's exactly at the limit
|
|
assert len(stacktrace4.encode("utf-8")) == MAX_OUTPUT_LEN
|
|
assert "...truncated" not in stacktrace4
|
|
assert stacktrace4 == exact_size_content.decode("utf-8", errors="ignore")
|
|
|
|
# Test case 5: Just over the limit (1MB + 1 byte)
|
|
over_limit_content = (
|
|
b"INFO: Seed: 12345\n" + b"C" * (MAX_OUTPUT_LEN - 19) + b"\n==ERROR: AddressSanitizer: heap-buffer-overflow"
|
|
)
|
|
|
|
result5 = ReproduceResult(
|
|
command_result=CommandResult(success=False, returncode=1, output=over_limit_content, error=None),
|
|
)
|
|
stacktrace5 = result5.stacktrace()
|
|
assert stacktrace5 is not None
|
|
|
|
# Should be truncated since it's over the limit
|
|
assert len(stacktrace5.encode("utf-8")) <= MAX_OUTPUT_LEN + 100
|
|
assert "...truncated" in stacktrace5
|
|
|
|
# Test case 6: Very large string (2MB) to test extreme truncation
|
|
very_large_content = (
|
|
b"INFO: Seed: 12345\n" + b"D" * (2 * MAX_OUTPUT_LEN) + b"\n==ERROR: AddressSanitizer: heap-buffer-overflow"
|
|
)
|
|
|
|
result6 = ReproduceResult(
|
|
command_result=CommandResult(success=False, returncode=1, output=very_large_content, error=None),
|
|
)
|
|
stacktrace6 = result6.stacktrace()
|
|
assert stacktrace6 is not None
|
|
|
|
# Should be significantly truncated
|
|
assert len(stacktrace6.encode("utf-8")) <= MAX_OUTPUT_LEN + 100
|
|
assert "...truncated" in stacktrace6
|
|
|
|
# Verify the truncation marker shows the correct number of truncated bytes
|
|
truncation_info = stacktrace6.split("...truncated")[1].split(" bytes...")[0]
|
|
truncated_bytes = int(truncation_info)
|
|
assert truncated_bytes >= MAX_OUTPUT_LEN # Should have truncated at least 1MB
|
|
|
|
|
|
def test_reproduce_result_methods():
|
|
"""Test the did_run and did_crash methods of ReproduceResult."""
|
|
# Test case 1: Successful run, no crash
|
|
result1 = ReproduceResult(
|
|
command_result=CommandResult(
|
|
success=True,
|
|
returncode=0,
|
|
output=b"INFO: Seed: 12345\nRunning normally",
|
|
error=None,
|
|
),
|
|
)
|
|
assert result1.did_run() is True
|
|
assert result1.did_crash() is False
|
|
|
|
# Test case 2: Failed run, no crash (fuzzer didn't start)
|
|
result2 = ReproduceResult(
|
|
command_result=CommandResult(success=False, returncode=1, output=b"Error: Could not start fuzzer", error=None),
|
|
)
|
|
assert result2.did_run() is False
|
|
assert result2.did_crash() is False
|
|
|
|
# Test case 3: Successful run with crash
|
|
result3 = ReproduceResult(
|
|
command_result=CommandResult(
|
|
success=False,
|
|
returncode=1,
|
|
output=b"INFO: Seed: 12345\nRunning normally\n==ERROR: AddressSanitizer: heap-buffer-overflow",
|
|
error=None,
|
|
),
|
|
)
|
|
assert result3.did_run() is True
|
|
assert result3.did_crash() is True
|
|
|
|
# Test case 4: Run with seed info in error output
|
|
result4 = ReproduceResult(
|
|
command_result=CommandResult(
|
|
success=False,
|
|
returncode=1,
|
|
output=None,
|
|
error=b"INFO: Seed: 12345\nRunning normally\n==ERROR: AddressSanitizer: heap-buffer-overflow",
|
|
),
|
|
)
|
|
assert result4.did_run() is True
|
|
assert result4.did_crash() is True
|
|
|
|
# Test case 5: Run with None returncode
|
|
result5 = ReproduceResult(
|
|
command_result=CommandResult(
|
|
success=False,
|
|
returncode=None,
|
|
output=b"INFO: Seed: 12345\nRunning normally",
|
|
error=None,
|
|
),
|
|
)
|
|
assert result5.did_run() is True
|
|
assert result5.did_crash() is False
|
|
|
|
# Test case 6: Timeout with crash token (should be detected as crash)
|
|
result6 = ReproduceResult(
|
|
command_result=CommandResult(
|
|
success=False,
|
|
returncode=124, # TIMEOUT_ERR_RESULT
|
|
output=b"INFO: Seed: 12345\nRunning normally\n==ERROR: AddressSanitizer: heap-buffer-overflow\nTimeout occurred",
|
|
error=None,
|
|
),
|
|
)
|
|
assert result6.did_run() is True
|
|
assert result6.did_crash() is True # Should detect crash due to crash token in stacktrace
|
|
|
|
# Test case 7: Timeout without crash token (should not be detected as crash)
|
|
result7 = ReproduceResult(
|
|
command_result=CommandResult(
|
|
success=False,
|
|
returncode=124, # TIMEOUT_ERR_RESULT
|
|
output=b"INFO: Seed: 12345\nRunning normally\nTimeout occurred\nNo crash detected",
|
|
error=None,
|
|
),
|
|
)
|
|
assert result7.did_run() is True
|
|
assert result7.did_crash() is False # Should not detect crash due to no crash token
|
|
|
|
# Test case 8: Timeout with crash token in error output
|
|
output = b"""/out/fuzzer-postauth_nomaths -rss_limit_mb=2560 -timeout=25 -runs=100 /testcase -max_len=50000 -timeout_exitcode=0 -dict=fuzzer-postauth_nomaths.dict < /dev/null
|
|
INFO: found LLVMFuzzerCustomMutator (0x5595fa311560). Disabling -len_control by default.
|
|
Dictionary: 58 entries
|
|
INFO: Running with entropic power schedule (0xFF, 100).
|
|
INFO: Seed: 3932698478
|
|
INFO: Loaded 1 modules (7459 inline 8-bit counters): 7459 [0x5595fa3cec00, 0x5595fa3d0923),
|
|
INFO: Loaded 1 PC tables (7459 PCs): 7459 [0x5595fa3d0928,0x5595fa3edb58),
|
|
/out/fuzzer-postauth_nomaths: Running 1 inputs 100 time(s) each.
|
|
Running: /testcase
|
|
Dropbear fuzzer: Disabling stderr output
|
|
fuzzer-postauth_nomaths: src/../fuzz/fuzz-wrapfd.c:216: int wrapfd_select(int, fd_set *, fd_set *, fd_set *, struct timeval *): Assertion `wrap_fds[i].mode != UNUSED' failed.
|
|
AddressSanitizer:DEADLYSIGNAL
|
|
=================================================================
|
|
==18==ERROR: AddressSanitizer: ABRT on unknown address 0x000000000012 (pc 0x7f99ebd7600b bp 0x7f99ebeeb588 sp 0x7ffe363d6470 T0)
|
|
SCARINESS: 10 (signal)
|
|
#0 0x7f99ebd7600b in raise (/lib/x86_64-linux-gnu/libc.so.6+0x4300b) (BuildId: 5792732f783158c66fb4f3756458ca24e46e827d)
|
|
#1 0x7f99ebd55858 in abort (/lib/x86_64-linux-gnu/libc.so.6+0x22858) (BuildId: 5792732f783158c66fb4f3756458ca24e46e827d)
|
|
#2 0x7f99ebd55728 (/lib/x86_64-linux-gnu/libc.so.6+0x22728) (BuildId: 5792732f783158c66fb4f3756458ca24e46e827d)
|
|
#3 0x7f99ebd66fd5 in __assert_fail (/lib/x86_64-linux-gnu/libc.so.6+0x33fd5) (BuildId: 5792732f783158c66fb4f3756458ca24e46e827d)
|
|
#4 0x5595fa2befa5 in wrapfd_select /src/dropbear/src/../fuzz/fuzz-wrapfd.c:216:5
|
|
#5 0x5595fa2c004d in session_loop /src/dropbear/src/common-session.c:210:9
|
|
#6 0x5595fa3064a0 in svr_session /src/dropbear/src/svr-session.c:208:2
|
|
#7 0x5595fa2bcff1 in fuzz_run_server /src/dropbear/src/../fuzz/fuzz-common.c:287:9
|
|
#8 0x5595fa156620 in fuzzer::Fuzzer::ExecuteCallback(unsigned char const*, unsigned long) /src/llvm-project/compiler-rt/lib/fuzzer/FuzzerLoop.cpp:614:13
|
|
#9 0x5595fa141895 in fuzzer::RunOneTest(fuzzer::Fuzzer*, char const*, unsigned long) /src/llvm-project/compiler-rt/lib/fuzzer/FuzzerDriver.cpp:327:6
|
|
#10 0x5595fa14732f in fuzzer::FuzzerDriver(int*, char***, int (*)(unsigned char const*, unsigned long)) /src/llvm-project/compiler-rt/lib/fuzzer/FuzzerDriver.cpp:862:9
|
|
#11 0x5595fa1725d2 in main /src/llvm-project/compiler-rt/lib/fuzzer/FuzzerMain.cpp:20:10
|
|
#12 0x7f99ebd57082 in __libc_start_main (/lib/x86_64-linux-gnu/libc.so.6+0x24082) (BuildId: 5792732f783158c66fb4f3756458ca24e46e827d)
|
|
#13 0x5595fa139a7d in _start (/out/fuzzer-postauth_nomaths+0x7ea7d)
|
|
|
|
DEDUP_TOKEN: raise--abort--
|
|
AddressSanitizer can not provide additional info.
|
|
SUMMARY: AddressSanitizer: ABRT (/lib/x86_64-linux-gnu/libc.so.6+0x4300b) (BuildId: 5792732f783158c66fb4f3756458ca24e46e827d) in raise
|
|
==18==ABORTING
|
|
MS: 0 ; base unit: 0000000000000000000000000000000000000000
|
|
subprocess command returned a non-zero exit status: 1"""
|
|
result8 = ReproduceResult(
|
|
command_result=CommandResult(
|
|
success=False,
|
|
returncode=124, # TIMEOUT_ERR_RESULT
|
|
output=b"INFO: Seed: 12345\nRunning normally\n" + output + b"\nTimeout occurred",
|
|
error=b"",
|
|
),
|
|
)
|
|
assert result8.did_run() is True
|
|
assert result8.did_crash() is True # Should detect crash due to crash token in error output
|
|
|
|
# Test case 9: Timeout with empty output (should not be detected as crash)
|
|
result9 = ReproduceResult(
|
|
command_result=CommandResult(
|
|
success=False,
|
|
returncode=124, # TIMEOUT_ERR_RESULT
|
|
output=None,
|
|
error=None,
|
|
),
|
|
)
|
|
assert result9.did_run() is False
|
|
assert result9.did_crash() is False # Should not detect crash due to no output and no crash token
|
|
|
|
# Test case 10: Timeout with only seed info (should not be detected as crash)
|
|
result10 = ReproduceResult(
|
|
command_result=CommandResult(
|
|
success=False,
|
|
returncode=124, # TIMEOUT_ERR_RESULT
|
|
output=b"INFO: Seed: 12345\nTimeout occurred",
|
|
error=None,
|
|
),
|
|
)
|
|
assert result10.did_run() is True
|
|
assert result10.did_crash() is False # Should not detect crash due to no crash token
|
|
|
|
# Test case 11: FAILURE_ERR_RESULT (201) - should not be detected as crash
|
|
result11 = ReproduceResult(
|
|
command_result=CommandResult(
|
|
success=False,
|
|
returncode=201, # FAILURE_ERR_RESULT
|
|
output=b"INFO: Seed: 12345\nRunning normally\n==ERROR: AddressSanitizer: heap-buffer-overflow",
|
|
error=None,
|
|
),
|
|
)
|
|
assert result11.did_run() is True
|
|
assert result11.did_crash() is False # Should not detect crash due to FAILURE_ERR_RESULT
|
|
|
|
# Test case 12: Different crash patterns in timeout scenario
|
|
result12 = ReproduceResult(
|
|
command_result=CommandResult(
|
|
success=False,
|
|
returncode=124, # TIMEOUT_ERR_RESULT
|
|
output=b"INFO: Seed: 12345\nRunning normally\n" + output + b"\nTimeout occurred",
|
|
error=None,
|
|
),
|
|
)
|
|
assert result12.did_run() is True
|
|
assert result12.did_crash() is True # Should detect crash due to UBSan error in stacktrace
|
|
|
|
# Test case 13: Timeout with multiple crash patterns
|
|
result13 = ReproduceResult(
|
|
command_result=CommandResult(
|
|
success=False,
|
|
returncode=124, # TIMEOUT_ERR_RESULT
|
|
output=b"INFO: Seed: 12345\nRunning normally\n" + output + b"\n" + output + b"\nTimeout occurred",
|
|
error=None,
|
|
),
|
|
)
|
|
assert result13.did_run() is True
|
|
assert result13.did_crash() is True # Should detect crash due to multiple crash patterns in stacktrace
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_exec_docker_cmd_grep_after_build(libjpeg_oss_fuzz_task_rw: ChallengeTask):
|
|
"""Test exec_docker_cmd running grep after build_fuzzers has been called."""
|
|
# First call build_fuzzers to set up container
|
|
libjpeg_oss_fuzz_task_rw.build_fuzzers()
|
|
|
|
# Then test grep command
|
|
result = libjpeg_oss_fuzz_task_rw.exec_docker_cmd(["ls", "-lah", "wrjpgcom"])
|
|
|
|
assert result.success is True
|
|
assert result.returncode == 0
|
|
assert b"-rwxr-xr-x" in result.output and b"wrjpgcom" in result.output
|
|
|
|
|
|
def test_apply_patch_diff_with_file(challenge_task: ChallengeTask):
|
|
"""Test applying a specific patch diff file to the source code."""
|
|
# Create a test file to patch
|
|
test_file = challenge_task.get_source_path() / "test_file.txt"
|
|
test_file.write_text("original content\n")
|
|
|
|
# Create a git diff patch file
|
|
diff_file = challenge_task.get_diff_path() / "test_patch.diff"
|
|
diff_content = """diff --git a/test_file.txt b/test_file.txt
|
|
index 1234567..abcdefg 100644
|
|
--- a/test_file.txt
|
|
+++ b/test_file.txt
|
|
@@ -1 +1,2 @@
|
|
-original content
|
|
+modified content
|
|
+new line
|
|
"""
|
|
diff_file.write_text(diff_content)
|
|
|
|
# Apply the specific patch file
|
|
result = challenge_task.apply_patch_diff(diff_file=diff_file)
|
|
|
|
assert result is True
|
|
assert test_file.exists()
|
|
assert test_file.read_text() == "modified content\nnew line\n"
|
|
|
|
|
|
def test_apply_patch_diff_binary_file(challenge_task: ChallengeTask):
|
|
"""Test applying a git diff patch for a binary file."""
|
|
# Create a binary test file
|
|
test_binary = challenge_task.get_source_path() / "test.bin"
|
|
exp_test_binary_content = """UEsDBBQAAAgIAHVmrlgtOwivDgAAAAwAAAAQADUAJTc0JTY1JTczJTc0LnR4dFVUDQAHnpZDZp6W
|
|
Q2aelkNmCgAgAAAAAAABABgAYaCn/x6m2gFhoKf/HqbaAWGgp/8eptoBy0jNyclXKM8vyknhAgBQ
|
|
SwECFAAUAAAICAB1Zq5YLTsIrw4AAAAMAAAAEAAtAAAAAAAAAAAAAAAAAAAAJTc0JTY1JTczJTc0
|
|
LnR4dFVUBQAHnpZDZgoAIAAAAAAAAQAYAGGgp/8eptoBYaCn/x6m2gFhoKf/HqbaAVBLBQYAAAAA
|
|
AQABAGsAAABxAAAAAAA="""
|
|
|
|
# Create a git diff patch file for binary
|
|
diff_file = challenge_task.get_diff_path() / "binary_patch.diff"
|
|
diff_content = """diff --git a/test.bin b/test.bin
|
|
new file mode 100644
|
|
index 0000000000000000000000000000000000000000..e3ec044f087ad576d0dffefc4ec71cd17011c287
|
|
GIT binary patch
|
|
literal 242
|
|
zcmWIWW@Zs#VBp|jC{0@zp=-^to{xcnfd_~M7)%*d%}rFzOjXT|fegKpijvR}UIzAg
|
|
z)11>_n2SLHsFZ<$kwJnXal!Kca?5TpqSL26&YnCOu5n)fl;=Yxh5&CyCJ_c)R_cOH
|
|
dLlBq_V1+n<7>fhES=k_tV`Rt%G77=w0sx8GJ7xd?
|
|
|
|
literal 0
|
|
HcmV?d00001
|
|
|
|
"""
|
|
diff_file.write_text(diff_content)
|
|
|
|
# Apply the binary patch
|
|
result = challenge_task.apply_patch_diff(diff_file=diff_file)
|
|
|
|
assert result is True
|
|
assert base64.b64decode(exp_test_binary_content) == test_binary.read_bytes()
|
|
|
|
|
|
def test_apply_patch_diff_file_not_found(challenge_task: ChallengeTask):
|
|
"""Test applying a patch file that doesn't exist."""
|
|
non_existent_diff = challenge_task.get_diff_path() / "nonexistent.diff"
|
|
|
|
with pytest.raises(ChallengeTaskError, match="Diff file .* not found"):
|
|
challenge_task.apply_patch_diff(diff_file=non_existent_diff)
|
|
|
|
|
|
def test_apply_patch_diff_git_apply_failure(challenge_task: ChallengeTask):
|
|
"""Test applying a patch file that causes git apply to fail."""
|
|
# Create a test file
|
|
test_file = challenge_task.get_source_path() / "test_file.txt"
|
|
test_file.write_text("original content\n")
|
|
|
|
# Create an invalid diff file
|
|
diff_file = challenge_task.get_diff_path() / "invalid_patch.diff"
|
|
diff_file.write_text("invalid diff content")
|
|
|
|
# Mock subprocess to simulate git apply failure
|
|
with patch("subprocess.run") as mock_run:
|
|
# Simulate git apply failure
|
|
mock_run.side_effect = subprocess.CalledProcessError(
|
|
returncode=1,
|
|
cmd=["patch", "-p1"],
|
|
output="",
|
|
stderr="patch does not apply",
|
|
)
|
|
|
|
with pytest.raises(ChallengeTaskError, match="Error applying diff"):
|
|
challenge_task.apply_patch_diff(diff_file=diff_file)
|
|
|
|
|
|
def test_oss_fuzz_container_org_gcr_io_oss_fuzz(challenge_task: ChallengeTask, tmp_path: Path):
|
|
"""Test oss_fuzz_container_org with gcr.io/oss-fuzz BASE_RUNNER_IMAGE."""
|
|
# Create a mock helper.py file with gcr.io/oss-fuzz BASE_RUNNER_IMAGE
|
|
oss_fuzz_path = challenge_task.get_oss_fuzz_path()
|
|
helper_file = oss_fuzz_path / "infra" / "helper.py"
|
|
helper_file.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
helper_content = """
|
|
import os
|
|
import sys
|
|
|
|
BASE_RUNNER_IMAGE = 'gcr.io/oss-fuzz-base/base-runner'
|
|
BASE_IMAGE_TAG = 'v1.0.0'
|
|
|
|
def main():
|
|
pass
|
|
"""
|
|
helper_file.write_text(helper_content)
|
|
|
|
# Mock the _helper_path to point to our created file
|
|
challenge_task._helper_path = helper_file
|
|
|
|
assert challenge_task.oss_fuzz_container_org == "gcr.io/oss-fuzz"
|
|
|
|
|
|
def test_oss_fuzz_container_org_aixcc_finals(challenge_task: ChallengeTask, tmp_path: Path):
|
|
"""Test oss_fuzz_container_org with ghcr.io/aixcc-finals BASE_RUNNER_IMAGE."""
|
|
# Create a mock helper.py file with ghcr.io/aixcc-finals BASE_RUNNER_IMAGE
|
|
oss_fuzz_path = challenge_task.get_oss_fuzz_path()
|
|
helper_file = oss_fuzz_path / "infra" / "helper.py"
|
|
helper_file.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
helper_content = """
|
|
import os
|
|
import sys
|
|
|
|
BASE_RUNNER_IMAGE = 'ghcr.io/aixcc-finals/base-runner'
|
|
BASE_IMAGE_TAG = 'v1.0.0'
|
|
|
|
def main():
|
|
pass
|
|
"""
|
|
helper_file.write_text(helper_content)
|
|
|
|
# Mock the _helper_path to point to our created file
|
|
challenge_task._helper_path = helper_file
|
|
|
|
assert challenge_task.oss_fuzz_container_org == "aixcc-afc"
|
|
|
|
|
|
def test_oss_fuzz_container_org_unknown_image(challenge_task: ChallengeTask, tmp_path: Path):
|
|
"""Test oss_fuzz_container_org with unknown BASE_RUNNER_IMAGE returns default."""
|
|
# Create a mock helper.py file with unknown BASE_RUNNER_IMAGE
|
|
oss_fuzz_path = challenge_task.get_oss_fuzz_path()
|
|
helper_file = oss_fuzz_path / "infra" / "helper.py"
|
|
helper_file.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
helper_content = """
|
|
import os
|
|
import sys
|
|
|
|
BASE_RUNNER_IMAGE = 'docker.io/unknown/registry/base-runner'
|
|
BASE_IMAGE_TAG = 'v1.0.0'
|
|
|
|
def main():
|
|
pass
|
|
"""
|
|
helper_file.write_text(helper_content)
|
|
|
|
# Mock the _helper_path to point to our created file
|
|
challenge_task._helper_path = helper_file
|
|
|
|
# Should return default when image doesn't match known patterns
|
|
assert challenge_task.oss_fuzz_container_org == "gcr.io/oss-fuzz"
|
|
|
|
|
|
def test_oss_fuzz_container_org_no_base_runner_image(challenge_task: ChallengeTask, tmp_path: Path):
|
|
"""Test oss_fuzz_container_org when BASE_RUNNER_IMAGE is not present."""
|
|
# Create a mock helper.py file without BASE_RUNNER_IMAGE
|
|
oss_fuzz_path = challenge_task.get_oss_fuzz_path()
|
|
helper_file = oss_fuzz_path / "infra" / "helper.py"
|
|
helper_file.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
helper_content = """
|
|
import os
|
|
import sys
|
|
|
|
BASE_IMAGE_TAG = 'v1.0.0'
|
|
|
|
def main():
|
|
pass
|
|
"""
|
|
helper_file.write_text(helper_content)
|
|
|
|
# Mock the _helper_path to point to our created file
|
|
challenge_task._helper_path = helper_file
|
|
|
|
# Should return default when BASE_RUNNER_IMAGE is not found
|
|
assert challenge_task.oss_fuzz_container_org == "gcr.io/oss-fuzz"
|
|
|
|
|
|
def test_oss_fuzz_container_org_malformed_line(challenge_task: ChallengeTask, tmp_path: Path):
|
|
"""Test oss_fuzz_container_org with malformed BASE_RUNNER_IMAGE line."""
|
|
# Create a mock helper.py file with malformed BASE_RUNNER_IMAGE line
|
|
oss_fuzz_path = challenge_task.get_oss_fuzz_path()
|
|
helper_file = oss_fuzz_path / "infra" / "helper.py"
|
|
helper_file.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
helper_content = """
|
|
import os
|
|
import sys
|
|
|
|
BASE_RUNNER_IMAGE = gcr.io/oss-fuzz-base/base-runner # Missing quotes
|
|
BASE_IMAGE_TAG = 'v1.0.0'
|
|
|
|
def main():
|
|
pass
|
|
"""
|
|
helper_file.write_text(helper_content)
|
|
|
|
# Mock the _helper_path to point to our created file
|
|
challenge_task._helper_path = helper_file
|
|
|
|
# Should return default when regex doesn't match
|
|
assert challenge_task.oss_fuzz_container_org == "gcr.io/oss-fuzz"
|
|
|
|
|
|
def test_oss_fuzz_container_org_multiple_matches(challenge_task: ChallengeTask, tmp_path: Path):
|
|
"""Test oss_fuzz_container_org with multiple BASE_RUNNER_IMAGE lines (should use first match)."""
|
|
# Create a mock helper.py file with multiple BASE_RUNNER_IMAGE lines
|
|
oss_fuzz_path = challenge_task.get_oss_fuzz_path()
|
|
helper_file = oss_fuzz_path / "infra" / "helper.py"
|
|
helper_file.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
helper_content = """
|
|
import os
|
|
import sys
|
|
|
|
BASE_RUNNER_IMAGE = 'ghcr.io/aixcc-finals/base-runner'
|
|
BASE_IMAGE_TAG = 'v1.0.0'
|
|
|
|
BASE_RUNNER_IMAGE = 'gcr.io/oss-fuzz-base/base-runner'
|
|
|
|
def main():
|
|
pass
|
|
"""
|
|
helper_file.write_text(helper_content)
|
|
|
|
# Mock the _helper_path to point to our created file
|
|
challenge_task._helper_path = helper_file
|
|
|
|
# Should use the last match
|
|
assert challenge_task.oss_fuzz_container_org == "gcr.io/oss-fuzz"
|
|
|
|
|
|
def test_oss_fuzz_container_org_multiple_matches_with_comment(challenge_task: ChallengeTask, tmp_path: Path):
|
|
"""Test oss_fuzz_container_org with multiple BASE_RUNNER_IMAGE lines (should use first match)."""
|
|
# Create a mock helper.py file with multiple BASE_RUNNER_IMAGE lines
|
|
oss_fuzz_path = challenge_task.get_oss_fuzz_path()
|
|
helper_file = oss_fuzz_path / "infra" / "helper.py"
|
|
helper_file.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
helper_content = """
|
|
import os
|
|
import sys
|
|
|
|
BASE_RUNNER_IMAGE = 'ghcr.io/aixcc-finals/base-runner'
|
|
BASE_IMAGE_TAG = 'v1.0.0'
|
|
|
|
# Comment mentioning BASE_RUNNER_IMAGE
|
|
# BASE_RUNNER_IMAGE = 'gcr.io/oss-fuzz-base/base-runner'
|
|
|
|
def main():
|
|
pass
|
|
"""
|
|
helper_file.write_text(helper_content)
|
|
|
|
# Mock the _helper_path to point to our created file
|
|
challenge_task._helper_path = helper_file
|
|
|
|
# Should use the first match (aixcc-finals)
|
|
assert challenge_task.oss_fuzz_container_org == "aixcc-afc"
|
|
|
|
|
|
def test_oss_fuzz_container_org_with_spaces(challenge_task: ChallengeTask, tmp_path: Path):
|
|
"""Test oss_fuzz_container_org with BASE_RUNNER_IMAGE line containing extra spaces."""
|
|
# Create a mock helper.py file with BASE_RUNNER_IMAGE line containing extra spaces
|
|
oss_fuzz_path = challenge_task.get_oss_fuzz_path()
|
|
helper_file = oss_fuzz_path / "infra" / "helper.py"
|
|
helper_file.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
helper_content = """
|
|
import os
|
|
import sys
|
|
|
|
BASE_RUNNER_IMAGE = 'ghcr.io/aixcc-finals/base-runner'
|
|
BASE_IMAGE_TAG = 'v1.0.0'
|
|
|
|
def main():
|
|
pass
|
|
"""
|
|
helper_file.write_text(helper_content)
|
|
|
|
# Mock the _helper_path to point to our created file
|
|
challenge_task._helper_path = helper_file
|
|
|
|
# Should handle extra spaces correctly
|
|
assert challenge_task.oss_fuzz_container_org == "aixcc-afc"
|
|
|
|
|
|
def test_oss_fuzz_container_org_with_double_quotes(challenge_task: ChallengeTask, tmp_path: Path):
|
|
"""Test oss_fuzz_container_org with BASE_RUNNER_IMAGE using double quotes."""
|
|
# Create a mock helper.py file with BASE_RUNNER_IMAGE using double quotes
|
|
oss_fuzz_path = challenge_task.get_oss_fuzz_path()
|
|
helper_file = oss_fuzz_path / "infra" / "helper.py"
|
|
helper_file.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
helper_content = """
|
|
import os
|
|
import sys
|
|
|
|
BASE_RUNNER_IMAGE = "ghcr.io/aixcc-finals/base-runner"
|
|
BASE_IMAGE_TAG = 'v1.0.0'
|
|
|
|
def main():
|
|
pass
|
|
"""
|
|
helper_file.write_text(helper_content)
|
|
|
|
# Mock the _helper_path to point to our created file
|
|
challenge_task._helper_path = helper_file
|
|
|
|
# Should handle double quotes correctly
|
|
assert challenge_task.oss_fuzz_container_org == "aixcc-afc"
|
|
|
|
|
|
def test_oss_fuzz_container_org_with_format_string(challenge_task: ChallengeTask, tmp_path: Path):
|
|
"""Test oss_fuzz_container_org with BASE_RUNNER_IMAGE using format string."""
|
|
# Create a mock helper.py file with BASE_RUNNER_IMAGE using format string
|
|
oss_fuzz_path = challenge_task.get_oss_fuzz_path()
|
|
helper_file = oss_fuzz_path / "infra" / "helper.py"
|
|
helper_file.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
helper_content = """
|
|
import os
|
|
import sys
|
|
|
|
BASE_RUNNER_IMAGE = f"ghcr.io/aixcc-finals/base-runner:v{{BASE_IMAGE_TAG}}"
|
|
BASE_IMAGE_TAG = 'v1.0.0'
|
|
|
|
def main():
|
|
pass
|
|
"""
|
|
helper_file.write_text(helper_content)
|
|
|
|
# Mock the _helper_path to point to our created file
|
|
challenge_task._helper_path = helper_file
|
|
|
|
# Should handle double quotes correctly
|
|
assert challenge_task.oss_fuzz_container_org == "aixcc-afc"
|
|
|
|
|
|
def test_oss_fuzz_container_org_file_not_found(challenge_task: ChallengeTask, tmp_path: Path):
|
|
"""Test oss_fuzz_container_org when helper.py file doesn't exist."""
|
|
# Create a non-existent helper.py file path
|
|
oss_fuzz_path = challenge_task.get_oss_fuzz_path()
|
|
helper_file = oss_fuzz_path / "infra" / "nonexistent_helper.py"
|
|
|
|
# Mock the _helper_path to point to non-existent file
|
|
challenge_task._helper_path = helper_file
|
|
|
|
# Should return default when file doesn't exist
|
|
assert challenge_task.oss_fuzz_container_org == "gcr.io/oss-fuzz"
|
|
|
|
|
|
def test_oss_fuzz_container_org_empty_file(challenge_task: ChallengeTask, tmp_path: Path):
|
|
"""Test oss_fuzz_container_org with empty helper.py file."""
|
|
# Create an empty helper.py file
|
|
oss_fuzz_path = challenge_task.get_oss_fuzz_path()
|
|
helper_file = oss_fuzz_path / "infra" / "helper.py"
|
|
helper_file.parent.mkdir(parents=True, exist_ok=True)
|
|
helper_file.write_text("")
|
|
|
|
# Mock the _helper_path to point to our created file
|
|
challenge_task._helper_path = helper_file
|
|
|
|
# Should return default when file is empty
|
|
assert challenge_task.oss_fuzz_container_org == "gcr.io/oss-fuzz"
|
|
|
|
|
|
def test_oss_fuzz_container_org_comment_only_file(challenge_task: ChallengeTask, tmp_path: Path):
|
|
"""Test oss_fuzz_container_org with helper.py file containing only comments."""
|
|
# Create a helper.py file with only comments
|
|
oss_fuzz_path = challenge_task.get_oss_fuzz_path()
|
|
helper_file = oss_fuzz_path / "infra" / "helper.py"
|
|
helper_file.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
helper_content = """
|
|
# This is a comment
|
|
# BASE_RUNNER_IMAGE = 'ghcr.io/aixcc-finals/base-runner'
|
|
# Another comment
|
|
"""
|
|
helper_file.write_text(helper_content)
|
|
|
|
# Mock the _helper_path to point to our created file
|
|
challenge_task._helper_path = helper_file
|
|
|
|
# Should return default when no BASE_RUNNER_IMAGE line is found
|
|
assert challenge_task.oss_fuzz_container_org == "gcr.io/oss-fuzz"
|
|
|
|
|
|
def test_oss_fuzz_container_org_partial_match(challenge_task: ChallengeTask, tmp_path: Path):
|
|
"""Test oss_fuzz_container_org with BASE_RUNNER_IMAGE that partially matches known patterns."""
|
|
# Create a mock helper.py file with BASE_RUNNER_IMAGE that partially matches
|
|
oss_fuzz_path = challenge_task.get_oss_fuzz_path()
|
|
helper_file = oss_fuzz_path / "infra" / "helper.py"
|
|
helper_file.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
helper_content = """
|
|
import os
|
|
import sys
|
|
|
|
BASE_RUNNER_IMAGE = 'gcr.io/oss-fuzz-something-else/base-runner'
|
|
BASE_IMAGE_TAG = 'v1.0.0'
|
|
|
|
def main():
|
|
pass
|
|
"""
|
|
helper_file.write_text(helper_content)
|
|
|
|
# Mock the _helper_path to point to our created file
|
|
challenge_task._helper_path = helper_file
|
|
|
|
# Should return default when image doesn't exactly match known patterns
|
|
assert challenge_task.oss_fuzz_container_org == "gcr.io/oss-fuzz"
|
|
|
|
|
|
def test_oss_fuzz_container_org_case_sensitive(challenge_task: ChallengeTask, tmp_path: Path):
|
|
"""Test oss_fuzz_container_org with case variations in BASE_RUNNER_IMAGE."""
|
|
# Create a mock helper.py file with case variations
|
|
oss_fuzz_path = challenge_task.get_oss_fuzz_path()
|
|
helper_file = oss_fuzz_path / "infra" / "helper.py"
|
|
helper_file.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
helper_content = """
|
|
import os
|
|
import sys
|
|
|
|
BASE_RUNNER_IMAGE = 'GCR.IO/OSS-FUZZ-BASE/BASE-RUNNER'
|
|
BASE_IMAGE_TAG = 'v1.0.0'
|
|
|
|
def main():
|
|
pass
|
|
"""
|
|
helper_file.write_text(helper_content)
|
|
|
|
# Mock the _helper_path to point to our created file
|
|
challenge_task._helper_path = helper_file
|
|
|
|
# Should return default when case doesn't match (case sensitive)
|
|
assert challenge_task.oss_fuzz_container_org == "gcr.io/oss-fuzz"
|