Drop support for directory as input

The output directory structure is expected to reflect the input structure.

Having a single input file and a single extract directory is simple.

However supporting directories as input has problems:
subdirectories should be there in the output, but we also create
directories for extraction, thus it is very easy to craft an input that
has a conflicting output (usually an unblob output is one such input).

Author: e3krisztian

.github/workflows/build-publish-image.yml

@@ -52,7 +52,7 @@ jobs:
         run: docker run --rm ${{ env.DOCKER_IMAGE }} --show-external-dependencies
 
       - name: Check unblob - run for a file with --verbose
-        run: docker run --rm -v "$(pwd)"/tests/integration/archive/zip/regular:/test ${{ env.DOCKER_IMAGE }} -v -e /tmp /test
+        run: docker run --rm -v "$(pwd)"/tests/integration/archive/zip/regular:/test ${{ env.DOCKER_IMAGE }} -v -e /tmp /test/__input__/apple.zip
 
       - name: Build and push
         if: ${{ github.event_name == 'push' && github.ref_name == 'main' }}

tests/test_cleanup.py

@@ -34,10 +34,8 @@ def wrapzip(filename: str, content: bytes) -> bytes:
 
 
 @pytest.fixture()
-def input_dir(tmp_path: Path):
-    input_dir = tmp_path / "input"
-    input_dir.mkdir()
-    return input_dir
+def input_file(tmp_path: Path):
+    return tmp_path / "input_file"
 
 
 @pytest.fixture()
@@ -47,39 +45,39 @@ def output_dir(tmp_path):
     return output_dir
 
 
-def test_remove_extracted_chunks(input_dir: Path, output_dir: Path):
-    (input_dir / "blob").write_bytes(ZIP_BYTES)
+def test_remove_extracted_chunks(input_file: Path, output_dir: Path):
+    input_file.write_bytes(ZIP_BYTES)
     config = ExtractionConfig(
         extract_root=output_dir,
         entropy_depth=0,
     )
 
-    all_reports = process_file(config, input_dir)
+    all_reports = process_file(config, input_file)
     assert list(output_dir.glob("**/*.zip")) == []
     check_result(all_reports)
 
 
-def test_keep_all_problematic_chunks(input_dir: Path, output_dir: Path):
-    (input_dir / "blob").write_bytes(DAMAGED_ZIP_BYTES)
+def test_keep_all_problematic_chunks(input_file: Path, output_dir: Path):
+    input_file.write_bytes(DAMAGED_ZIP_BYTES)
     config = ExtractionConfig(
         extract_root=output_dir,
         entropy_depth=0,
     )
 
-    all_reports = process_file(config, input_dir)
+    all_reports = process_file(config, input_file)
     # damaged zip file should not be removed
     assert all_reports.errors != [], "Unexpectedly no errors found!"
     assert list(output_dir.glob("**/*.zip"))
 
 
-def test_keep_all_unknown_chunks(input_dir: Path, output_dir: Path):
-    (input_dir / "blob").write_bytes(b"unknown1" + ZIP_BYTES + b"unknown2")
+def test_keep_all_unknown_chunks(input_file: Path, output_dir: Path):
+    input_file.write_bytes(b"unknown1" + ZIP_BYTES + b"unknown2")
     config = ExtractionConfig(
         extract_root=output_dir,
         entropy_depth=0,
     )
 
-    all_reports = process_file(config, input_dir)
+    all_reports = process_file(config, input_file)
     assert list(output_dir.glob("**/*.unknown"))
     check_result(all_reports)
 
@@ -93,13 +91,13 @@ def calculate_chunk(self, file: File, start_offset: int) -> ValidChunk:
         return ValidChunk(start_offset=start_offset, end_offset=start_offset + 1)
 
 
-def test_keep_chunks_with_null_extractor(input_dir: Path, output_dir: Path):
-    (input_dir / "blob").write_text("some text")
+def test_keep_chunks_with_null_extractor(input_file: Path, output_dir: Path):
+    input_file.write_bytes(b"some text")
     config = ExtractionConfig(
         extract_root=output_dir,
         entropy_depth=0,
         handlers=(_HandlerWithNullExtractor,),
     )
-    all_reports = process_file(config, input_dir)
+    all_reports = process_file(config, input_file)
     assert list(output_dir.glob("**/*.null"))
     check_result(all_reports)

tests/test_cli.py

@@ -127,10 +127,10 @@ def test_non_existing_file(tmp_path: Path):
     result = runner.invoke(unblob.cli.cli, ["--extract-dir", str(tmp_path), str(path)])
     assert result.exit_code == 2
     assert "Invalid value for 'FILE'" in result.output
-    assert f"Path '{str(path)}' does not exist" in result.output
+    assert f"File '{str(path)}' does not exist" in result.output
 
 
-def test_empty_dir_as_file(tmp_path: Path):
+def test_dir_for_file(tmp_path: Path):
     runner = CliRunner()
     out_path = tmp_path.joinpath("out")
     out_path.mkdir()
@@ -139,7 +139,7 @@ def test_empty_dir_as_file(tmp_path: Path):
     result = runner.invoke(
         unblob.cli.cli, ["--extract-dir", str(out_path), str(in_path)]
     )
-    assert result.exit_code == 0
+    assert result.exit_code != 0
 
 
 @pytest.mark.parametrize(
@@ -172,7 +172,8 @@ def test_archive_success(
         / "archive"
         / "zip"
         / "regular"
-        / "__input__/"
+        / "__input__"
+        / "apple.zip"
     )
     process_file_mock = mock.MagicMock()
     logger_config_mock = mock.MagicMock()
@@ -214,7 +215,8 @@ def test_keep_extracted_chunks(
         / "archive"
         / "zip"
         / "regular"
-        / "__input__/"
+        / "__input__"
+        / "apple.zip"
     )
     params = args + ["--extract-dir", str(tmp_path), str(in_path)]
 

tests/test_handlers.py

@@ -32,10 +32,11 @@
 def test_all_handlers(
     input_dir: Path, output_dir: Path, extraction_config: ExtractionConfig
 ):
-    all_reports = process_file(extraction_config, input_dir)
+    for input_file in input_dir.iterdir():
+        reports = process_file(extraction_config, input_file)
+        check_result(reports)
 
     check_output_is_the_same(output_dir, extraction_config.extract_root)
-    check_result(all_reports)
 
 
 @pytest.mark.parametrize(

unblob/cli.py

@@ -79,7 +79,7 @@ def __init__(
 @click.command(help=get_help_text())
 @click.argument(
     "file",
-    type=click.Path(path_type=Path, exists=True, resolve_path=True),
+    type=click.Path(path_type=Path, dir_okay=False, exists=True, resolve_path=True),
     required=True,
 )
 @click.option(

unblob/processing.py

@@ -68,6 +68,9 @@ class ExtractionConfig:
 
 @terminate_gracefully
 def process_file(config: ExtractionConfig, path: Path) -> ProcessResult:
+    if not path.is_file():
+        raise ValueError("path is not a file", path)
+
     task = Task(
         path=path,
         depth=0,