diff --git a/poetry.lock b/poetry.lock index b388e82dbd..d7fc51c2b6 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1530,7 +1530,7 @@ develop = false type = "git" url = "https://github.com/onekey-sec/unblob-native" reference = "landlock" -resolved_reference = "a6b718ffb7918287cb1270d120c342a3a571443b" +resolved_reference = "f2887857364cff26b65bea7d46fc539c08358301" [[package]] name = "urllib3" diff --git a/unblob/processing.py b/unblob/processing.py index 83eb9dc64d..c78aa59c52 100644 --- a/unblob/processing.py +++ b/unblob/processing.py @@ -1,5 +1,6 @@ import multiprocessing import shutil +import sys from operator import attrgetter from pathlib import Path from typing import Iterable, List, Optional, Sequence, Set, Tuple, Type, Union @@ -9,6 +10,7 @@ import plotext as plt from structlog import get_logger from unblob_native import math_tools as mt +from unblob_native.sandbox import AccessFS, restrict_access # type: ignore from unblob.handlers import BUILTIN_DIR_HANDLERS, BUILTIN_HANDLERS, Handlers @@ -111,6 +113,29 @@ def get_extract_dir_for(self, path: Path) -> Path: return extract_dir.expanduser().resolve() +def sandbox(extract_dir: Path, report_file: Optional[Path]): + restrictions = [ + AccessFS.read("/"), + AccessFS.read_write("/dev/shm"), # noqa: S108 + AccessFS.read_write(extract_dir.as_posix()), + AccessFS.make_dir(extract_dir.parent.as_posix()), + ] + + if report_file: + restrictions += [ + AccessFS.read_write(report_file), + AccessFS.make_reg(report_file.parent), + ] + + if "pytest" in sys.modules: + restrictions += [ + AccessFS.read_write("/tmp"), # noqa: S108 + AccessFS.read_write(Path(__file__).parent.parent.resolve().as_posix()), + ] + + restrict_access(*restrictions) + + @terminate_gracefully def process_file( config: ExtractionConfig, input_path: Path, report_file: Optional[Path] = None @@ -135,6 +160,10 @@ def process_file( ) return ProcessResult() + if not hasattr(process_file, "_sandboxed"): + sandbox(extract_dir, report_file) + process_file._sandboxed = True # noqa: SLF001 + process_result = _process_task(config, task) if not config.skip_extraction: