From c7dba8de1e4a1f571aa952f5e0db3410eb6d35c9 Mon Sep 17 00:00:00 2001 From: tastelikefeet <58414341+tastelikefeet@users.noreply.github.com> Date: Mon, 15 Jun 2026 10:35:50 +0800 Subject: [PATCH] Fix some security issues (#1737) --- modelscope/models/base/base_model.py | 4 +- .../abnormal_object_detection/mmdet_model.py | 3 +- .../annotator/annotator.py | 16 +- .../cv/face_detection/scrfd/scrfd_detect.py | 9 +- .../cv/image_classification/mmcls_model.py | 6 + .../panseg_model.py | 6 + .../semantic_seg_model.py | 6 + .../models/cv/object_detection/mmdet_model.py | 3 +- .../cv/salient_detection/salient_model.py | 3 +- modelscope/pipelines/base.py | 4 +- .../cv/human3d_animation_pipeline.py | 10 + .../cv/object_detection_3d_pipeline.py | 3 +- .../cv/tinynas_classification_pipeline.py | 10 +- .../cv/action_detection_mapper.py | 8 +- .../cv/controllable_image_generation.py | 11 +- .../preprocessors/cv/mmcls_preprocessor.py | 8 + modelscope/utils/config.py | 48 +++- .../test_human3d_animation_security.py | 100 +++++++ .../pipelines/test_remote_config_security.py | 261 ++++++++++++++++++ .../test_tinynas_classification_security.py | 95 +++++++ .../test_action_detection_mapper_security.py | 162 +++++++++++ 21 files changed, 752 insertions(+), 24 deletions(-) create mode 100644 tests/pipelines/test_human3d_animation_security.py create mode 100644 tests/pipelines/test_remote_config_security.py create mode 100644 tests/pipelines/test_tinynas_classification_security.py create mode 100644 tests/preprocessors/test_action_detection_mapper_security.py diff --git a/modelscope/models/base/base_model.py b/modelscope/models/base/base_model.py index 4a1b6494..e2a305cf 100644 --- a/modelscope/models/base/base_model.py +++ b/modelscope/models/base/base_model.py @@ -53,7 +53,9 @@ class Model(ABC): 'import extra libs or execute the code in the model repo, setting this to true ' 'means you trust the files in it.') if not check_model_from_owner_group(model_dir=model_dir): - assert self.trust_remote_code, info_str + # `raise` (not `assert`) so the gate also holds under `python -O`. + if not self.trust_remote_code: + raise RuntimeError(info_str) @abstractmethod def forward(self, *args, **kwargs) -> Dict[str, Any]: diff --git a/modelscope/models/cv/abnormal_object_detection/mmdet_model.py b/modelscope/models/cv/abnormal_object_detection/mmdet_model.py index d341b13a..4626022d 100644 --- a/modelscope/models/cv/abnormal_object_detection/mmdet_model.py +++ b/modelscope/models/cv/abnormal_object_detection/mmdet_model.py @@ -27,7 +27,8 @@ class AbnormalDetectionModel(TorchModel): model_path = osp.join(model_dir, ModelFile.TORCH_MODEL_FILE) config_path = osp.join(model_dir, 'mmcv_config.py') - config = Config.from_file(config_path) + config = Config.from_file( + config_path, trust_remote_code=self.trust_remote_code) config.model.pretrained = None self.model = build_detector( config.model, test_cfg=config.get('test_cfg')) diff --git a/modelscope/models/cv/controllable_image_generation/annotator/annotator.py b/modelscope/models/cv/controllable_image_generation/annotator/annotator.py index 12076080..eff53416 100644 --- a/modelscope/models/cv/controllable_image_generation/annotator/annotator.py +++ b/modelscope/models/cv/controllable_image_generation/annotator/annotator.py @@ -361,14 +361,22 @@ def show_result_pyplot(model, class SegformerDetector: - def __init__(self, annotator_ckpts_path, device='cuda'): + def __init__(self, + annotator_ckpts_path, + device='cuda', + trust_remote_code=False): + from modelscope.utils.config import \ + check_trust_remote_code_for_config modelpath = os.path.join( annotator_ckpts_path, 'segformer_mit-b4_512x512_160k_ade20k_20220620_112216-4fa4f58f.pth' ) - config_file = os.path.join( - annotator_ckpts_path.replace('ckpt/annotator/', ''), - 'config/config.py') + annotator_root = annotator_ckpts_path.replace('ckpt/annotator/', '') + config_file = os.path.join(annotator_root, 'config/config.py') + check_trust_remote_code_for_config( + config_file, + trust_remote_code=trust_remote_code, + model_dir=annotator_root) self.model = init_segmentor(config_file, modelpath).to(device) def __call__(self, img): diff --git a/modelscope/models/cv/face_detection/scrfd/scrfd_detect.py b/modelscope/models/cv/face_detection/scrfd/scrfd_detect.py index e382ec9a..fd45045d 100644 --- a/modelscope/models/cv/face_detection/scrfd/scrfd_detect.py +++ b/modelscope/models/cv/face_detection/scrfd/scrfd_detect.py @@ -39,8 +39,15 @@ class ScrfdDetect(TorchModel): from modelscope.models.cv.face_detection.scrfd.mmdet_patch.models.backbones import ResNetV1e from modelscope.models.cv.face_detection.scrfd.mmdet_patch.models.dense_heads import SCRFDHead from modelscope.models.cv.face_detection.scrfd.mmdet_patch.models.detectors import SCRFD + from modelscope.utils.config import \ + check_trust_remote_code_for_config cfg_file = kwargs.get('config_file', 'mmcv_scrfd.py') - cfg = Config.fromfile(osp.join(model_dir, cfg_file)) + cfg_path = osp.join(model_dir, cfg_file) + check_trust_remote_code_for_config( + cfg_path, + trust_remote_code=self.trust_remote_code, + model_dir=model_dir) + cfg = Config.fromfile(cfg_path) model_file = kwargs.get('model_file', ModelFile.TORCH_MODEL_BIN_FILE) ckpt_path = osp.join(model_dir, model_file) cfg.model.test_cfg.score_thr = kwargs.get('score_thr', 0.3) diff --git a/modelscope/models/cv/image_classification/mmcls_model.py b/modelscope/models/cv/image_classification/mmcls_model.py index bd37d3de..a038551d 100644 --- a/modelscope/models/cv/image_classification/mmcls_model.py +++ b/modelscope/models/cv/image_classification/mmcls_model.py @@ -16,12 +16,18 @@ class ClassificationModel(TorchModel): from mmcls.models import build_classifier import modelscope.models.cv.image_classification.backbones from modelscope.utils.hub import read_config + from modelscope.utils.config import \ + check_trust_remote_code_for_config super().__init__(model_dir) self.config_type = 'ms_config' mm_config = os.path.join(model_dir, 'config.py') if os.path.exists(mm_config): + check_trust_remote_code_for_config( + mm_config, + trust_remote_code=self.trust_remote_code, + model_dir=model_dir) cfg = mmcv.Config.fromfile(mm_config) cfg.model.pretrained = None self.cls_model = build_classifier(cfg.model) diff --git a/modelscope/models/cv/image_panoptic_segmentation/panseg_model.py b/modelscope/models/cv/image_panoptic_segmentation/panseg_model.py index f44c01e8..685d6ec1 100644 --- a/modelscope/models/cv/image_panoptic_segmentation/panseg_model.py +++ b/modelscope/models/cv/image_panoptic_segmentation/panseg_model.py @@ -20,9 +20,15 @@ class SwinLPanopticSegmentation(TorchModel): from mmcv.runner import load_checkpoint import mmcv from mmdet.models import build_detector + from modelscope.utils.config import \ + check_trust_remote_code_for_config config = osp.join(model_dir, 'config.py') + check_trust_remote_code_for_config( + config, + trust_remote_code=self.trust_remote_code, + model_dir=model_dir) cfg = mmcv.Config.fromfile(config) if 'pretrained' in cfg.model: cfg.model.pretrained = None diff --git a/modelscope/models/cv/image_semantic_segmentation/semantic_seg_model.py b/modelscope/models/cv/image_semantic_segmentation/semantic_seg_model.py index 455f29fb..6ff088c6 100644 --- a/modelscope/models/cv/image_semantic_segmentation/semantic_seg_model.py +++ b/modelscope/models/cv/image_semantic_segmentation/semantic_seg_model.py @@ -27,8 +27,14 @@ class SemanticSegmentation(TorchModel): from mmcv.runner import load_checkpoint import mmcv from mmdet.models import build_detector + from modelscope.utils.config import \ + check_trust_remote_code_for_config config = osp.join(model_dir, 'mmcv_config.py') + check_trust_remote_code_for_config( + config, + trust_remote_code=self.trust_remote_code, + model_dir=model_dir) cfg = mmcv.Config.fromfile(config) if 'pretrained' in cfg.model: cfg.model.pretrained = None diff --git a/modelscope/models/cv/object_detection/mmdet_model.py b/modelscope/models/cv/object_detection/mmdet_model.py index 485d440a..26b6f57f 100644 --- a/modelscope/models/cv/object_detection/mmdet_model.py +++ b/modelscope/models/cv/object_detection/mmdet_model.py @@ -31,7 +31,8 @@ class DetectionModel(TorchModel): model_path = osp.join(model_dir, ModelFile.TORCH_MODEL_FILE) config_path = osp.join(model_dir, 'mmcv_config.py') - config = Config.from_file(config_path) + config = Config.from_file( + config_path, trust_remote_code=self.trust_remote_code) config.model.pretrained = None self.model = build_detector(config.model) diff --git a/modelscope/models/cv/salient_detection/salient_model.py b/modelscope/models/cv/salient_detection/salient_model.py index e25166c8..5d4c5646 100644 --- a/modelscope/models/cv/salient_detection/salient_model.py +++ b/modelscope/models/cv/salient_detection/salient_model.py @@ -32,7 +32,8 @@ class SalientDetection(TorchModel): self.model = U2NET(3, 1) else: self.model = SENet(backbone_path=None, pretrained=False) - config = Config.from_file(config_path) + config = Config.from_file( + config_path, trust_remote_code=self.trust_remote_code) self.norm_mean = config.norm_mean self.norm_std = config.norm_std self.norm_size = config.norm_size diff --git a/modelscope/pipelines/base.py b/modelscope/pipelines/base.py index 115a3b95..f7d06efc 100644 --- a/modelscope/pipelines/base.py +++ b/modelscope/pipelines/base.py @@ -151,7 +151,9 @@ class Pipeline(ABC): 'import extra libs or execute the code in the model repo, setting this to true ' 'means you trust the files in it.') if not check_model_from_owner_group(model_dir=model_dir): - assert self.trust_remote_code, info_str + # `raise` (not `assert`) so the gate also holds under `python -O`. + if not self.trust_remote_code: + raise RuntimeError(info_str) def prepare_model(self): """ Place model on certain device for pytorch models before first inference diff --git a/modelscope/pipelines/cv/human3d_animation_pipeline.py b/modelscope/pipelines/cv/human3d_animation_pipeline.py index 4e5ab46d..683e499e 100644 --- a/modelscope/pipelines/cv/human3d_animation_pipeline.py +++ b/modelscope/pipelines/cv/human3d_animation_pipeline.py @@ -72,6 +72,16 @@ class Human3DAnimationPipeline(Pipeline): (case_name, action_name)) exec_path = os.path.join(self.model_dir, 'skinning.py') + # `skinning.py` ships inside the model repo and is executed via Blender; + # gate it behind trust_remote_code to prevent RCE from untrusted repos. + self.check_trust_remote_code( + info_str= + ('Human3DAnimationPipeline executes `skinning.py` from the model ' + 'repository via Blender, which can run arbitrary code. ' + 'Pass `trust_remote_code=True` to pipeline() to opt in if you ' + 'trust the model repository.'), + model_dir=self.model_dir) + cmd = f'{self.blender} -b -P {exec_path} -- --input {self.case_dir}' \ f' --gltf_path {gltf_path} --action {self.action}' os.system(cmd) diff --git a/modelscope/pipelines/cv/object_detection_3d_pipeline.py b/modelscope/pipelines/cv/object_detection_3d_pipeline.py index 640d76f6..fc4eab48 100644 --- a/modelscope/pipelines/cv/object_detection_3d_pipeline.py +++ b/modelscope/pipelines/cv/object_detection_3d_pipeline.py @@ -49,7 +49,8 @@ class ObjectDetection3DPipeline(Pipeline): """ super().__init__(model=model, **kwargs) config_path = osp.join(model, 'mmcv_depe.py') - self.cfg = Config.from_file(config_path) + self.cfg = Config.from_file( + config_path, trust_remote_code=self.trust_remote_code) if torch.cuda.is_available(): self.device = torch.device('cuda') else: diff --git a/modelscope/pipelines/cv/tinynas_classification_pipeline.py b/modelscope/pipelines/cv/tinynas_classification_pipeline.py index ca039eca..8190b63d 100644 --- a/modelscope/pipelines/cv/tinynas_classification_pipeline.py +++ b/modelscope/pipelines/cv/tinynas_classification_pipeline.py @@ -1,5 +1,6 @@ # Copyright (c) Alibaba, Inc. and its affiliates. +import ast import math import os.path as osp from typing import Any, Dict @@ -83,10 +84,11 @@ class TinynasClassificationPipeline(Pipeline): def postprocess(self, inputs: Dict[str, Any]) -> Dict[str, Any]: label_mapping_path = osp.join(self.path, 'label_map.txt') - f = open(label_mapping_path, encoding='utf-8') - content = f.read() - f.close() - label_dict = eval(content) + with open(label_mapping_path, encoding='utf-8') as f: + content = f.read() + # `label_map.txt` ships from a remote model repo; restrict parsing to + # plain literal containers so a malicious file cannot trigger RCE (#1668). + label_dict = ast.literal_eval(content) output_prob = torch.nn.functional.softmax(inputs['outputs'], dim=-1) score = torch.max(output_prob) diff --git a/modelscope/preprocessors/cv/action_detection_mapper.py b/modelscope/preprocessors/cv/action_detection_mapper.py index 9bb6d422..669e45dc 100644 --- a/modelscope/preprocessors/cv/action_detection_mapper.py +++ b/modelscope/preprocessors/cv/action_detection_mapper.py @@ -1,5 +1,6 @@ # Copyright (c) Alibaba, Inc. and its affiliates. +import ast import copy import random @@ -84,7 +85,12 @@ class VideoDetMapper: def _call(self, data_dict): video_name = data_dict['path:FILE'] if data_dict['actions'] is not None: - data_dict['actions'] = eval(data_dict['actions']) + actions = data_dict['actions'] + if isinstance(actions, bytes): + actions = actions.decode('utf-8') + if isinstance(actions, str): + actions = ast.literal_eval(actions) + data_dict['actions'] = actions else: data_dict['actions'] = [] diff --git a/modelscope/preprocessors/cv/controllable_image_generation.py b/modelscope/preprocessors/cv/controllable_image_generation.py index 054baca3..6fb77872 100644 --- a/modelscope/preprocessors/cv/controllable_image_generation.py +++ b/modelscope/preprocessors/cv/controllable_image_generation.py @@ -58,7 +58,7 @@ def resize_image(input_image, resolution): return img -def build_detector(control_type, model_path, device): +def build_detector(control_type, model_path, device, trust_remote_code=False): if control_type == 'scribble': detector = None elif control_type == 'canny': @@ -74,7 +74,8 @@ def build_detector(control_type, model_path, device): elif control_type == 'pose': detector = OpenposeDetector(model_path, device) elif control_type == 'seg': - detector = SegformerDetector(model_path, device) + detector = SegformerDetector( + model_path, device, trust_remote_code=trust_remote_code) elif control_type == 'fake_scribble': detector = HEDdetector(model_path, device) else: @@ -136,8 +137,10 @@ class ControllableImageGenerationPreprocessor(Preprocessor): def __init__(self, mode=ModeKeys.INFERENCE, *args, **kwargs): super().__init__(mode=ModeKeys.INFERENCE, *args, **kwargs) self.detector = build_detector( - kwargs.get('control_type', 'hed'), kwargs.get('model_path', None), - kwargs.get('device', 'cuda')) + kwargs.get('control_type', 'hed'), + kwargs.get('model_path', None), + kwargs.get('device', 'cuda'), + trust_remote_code=kwargs.get('trust_remote_code', False)) @type_assert(object, object) def __call__(self, data: input, **kwargs) -> Dict[str, Any]: diff --git a/modelscope/preprocessors/cv/mmcls_preprocessor.py b/modelscope/preprocessors/cv/mmcls_preprocessor.py index 36e7ac4d..b549ee09 100644 --- a/modelscope/preprocessors/cv/mmcls_preprocessor.py +++ b/modelscope/preprocessors/cv/mmcls_preprocessor.py @@ -38,11 +38,19 @@ class ImageClassificationMmcvPreprocessor(Preprocessor): import mmcv from mmcls.datasets.pipelines import Compose from modelscope.models.cv.image_classification.utils import preprocess_transform + from modelscope.utils.config import \ + check_trust_remote_code_for_config + # Preprocessor base does not carry trust_remote_code; read it directly. + trust_remote_code = kwargs.get('trust_remote_code', False) super().__init__(**kwargs) self.config_type = 'ms_config' mm_config = os.path.join(model_dir, 'config.py') if os.path.exists(mm_config): + check_trust_remote_code_for_config( + mm_config, + trust_remote_code=trust_remote_code, + model_dir=model_dir) cfg = mmcv.Config.fromfile(mm_config) cfg.model.pretrained = None config_type = 'mmcv_config' diff --git a/modelscope/utils/config.py b/modelscope/utils/config.py index 099dbf11..2b2920cb 100644 --- a/modelscope/utils/config.py +++ b/modelscope/utils/config.py @@ -28,6 +28,40 @@ DEPRECATION_KEY = '_deprecation_' RESERVED_KEYS = ['filename', 'text', 'pretty_text'] +def check_trust_remote_code_for_config(filename, + trust_remote_code: bool = False, + model_dir=None): + """Refuse to exec a `.py` config file that comes from an untrusted source. + + Loading a Python config (via `Config.from_file`, `mmcv.Config.fromfile`, + `mmseg.apis.init_segmentor`, etc.) imports the file as a module, which + runs any top-level code it contains. Anything that ultimately reads a + `.py` config from a remote model repo MUST gate that load with this + helper. JSON / YAML configs are passive data and pass through. + + Args: + filename: Path to the candidate config file. + trust_remote_code: Caller opt-in flag; pass through + ``self.trust_remote_code`` from ``Model`` / ``Pipeline`` / + ``Preprocessor`` callers. + model_dir: Repo root used by the owner-group check. Defaults to the + parent directory of ``filename``. + """ + if not str(filename).endswith('.py'): + return + from modelscope.utils.automodel_utils import check_model_from_owner_group + if model_dir is None: + model_dir = osp.dirname(osp.abspath(osp.expanduser(str(filename)))) + if check_model_from_owner_group(model_dir=model_dir): + return + if trust_remote_code: + return + raise RuntimeError( + f'Refusing to load Python config "{filename}": doing so would execute ' + 'code from the model repository. Pass `trust_remote_code=True` to opt ' + 'in if you trust the source.') + + class ConfigDict(addict.Dict): """ Dict which support get value through getattr @@ -82,7 +116,7 @@ class Config: """ @staticmethod - def _file2dict(filename): + def _file2dict(filename, trust_remote_code: bool = False, model_dir=None): filename = osp.abspath(osp.expanduser(filename)) if not osp.exists(filename): raise ValueError(f'File does not exists {filename}') @@ -90,6 +124,9 @@ class Config: if fileExtname not in ['.py', '.json', '.yaml', '.yml']: raise IOError('Only py/yml/yaml/json type are supported now!') + check_trust_remote_code_for_config( + filename, trust_remote_code=trust_remote_code, model_dir=model_dir) + with tempfile.TemporaryDirectory() as tmp_cfg_dir: tmp_cfg_file = tempfile.NamedTemporaryFile( dir=tmp_cfg_dir, suffix=fileExtname) @@ -126,10 +163,11 @@ class Config: return cfg_dict, cfg_text @staticmethod - def from_file(filename): + def from_file(filename, trust_remote_code: bool = False, model_dir=None): if isinstance(filename, Path): filename = str(filename) - cfg_dict, cfg_text = Config._file2dict(filename) + cfg_dict, cfg_text = Config._file2dict( + filename, trust_remote_code=trust_remote_code, model_dir=model_dir) return Config(cfg_dict, cfg_text=cfg_text, filename=filename) @staticmethod @@ -156,7 +194,9 @@ class Config: temp_file.write(cfg_str) # on windows, previous implementation cause error # see PR 1077 for details - cfg = Config.from_file(temp_file.name) + # `from_string` materializes a caller-provided in-process string into + # a tempfile; the threat model for `trust_remote_code` does not apply. + cfg = Config.from_file(temp_file.name, trust_remote_code=True) os.remove(temp_file.name) return cfg diff --git a/tests/pipelines/test_human3d_animation_security.py b/tests/pipelines/test_human3d_animation_security.py new file mode 100644 index 00000000..ca403a66 --- /dev/null +++ b/tests/pipelines/test_human3d_animation_security.py @@ -0,0 +1,100 @@ +# Copyright (c) Alibaba, Inc. and its affiliates. +"""Regression test for Issue #1673. + +Verifies that Human3DAnimationPipeline.gen_weights() refuses to execute the +remote `skinning.py` script unless either (a) the model is owned by a trusted +group (`iic` / `damo`), or (b) the user explicitly opted in with +`trust_remote_code=True`. +""" +import os +import shutil +import sys +import tempfile +import types +import unittest +from unittest import mock + +from modelscope.pipelines.cv.human3d_animation_pipeline import \ + Human3DAnimationPipeline # noqa: E402 + +# `modelscope.models.cv.human3d_animation` lazily resolves to symbols that +# pull in heavy optional deps (e.g. `nvdiffrast`). We never invoke those +# symbols inside `gen_weights`, so substitute a lightweight stub *before* the +# pipeline module is imported. Only stub if the real package cannot be +# resolved, so we don't shadow it for environments that do have the deps. +_PKG = 'modelscope.models.cv.human3d_animation' +if _PKG not in sys.modules or not hasattr(sys.modules[_PKG], + 'gen_skeleton_bvh'): + try: + from modelscope.models.cv.human3d_animation import ( # noqa: F401 + gen_skeleton_bvh, read_obj, write_obj, + ) + except Exception: + _stub = types.ModuleType(_PKG) + _stub.gen_skeleton_bvh = lambda *a, **kw: None + _stub.read_obj = lambda *a, **kw: None + _stub.write_obj = lambda *a, **kw: None + sys.modules[_PKG] = _stub + + +class Human3DAnimationSecurityTest(unittest.TestCase): + """Pin the trust_remote_code gate around remote `skinning.py` execution.""" + + def setUp(self): + self.tmp_root = tempfile.mkdtemp(prefix='ms_human3d_security_') + # `check_model_from_owner_group` extracts the owner from the parent + # directory name, so model dirs must live two levels deep. + self.untrusted_dir = os.path.join(self.tmp_root, 'attacker', 'badrepo') + self.trusted_dir = os.path.join(self.tmp_root, 'damo', 'goodrepo') + for d in (self.untrusted_dir, self.trusted_dir): + os.makedirs(d) + with open(os.path.join(d, 'skinning.py'), 'w') as f: + f.write("raise SystemExit('this script must not run')\n") + + self.case_dir = os.path.join(self.tmp_root, 'case') + os.makedirs(self.case_dir) + self.save_dir = os.path.join(self.tmp_root, 'out') + + def tearDown(self): + shutil.rmtree(self.tmp_root, ignore_errors=True) + + def _make_pipeline(self, model_dir, trust_remote_code): + # Bypass Pipeline.__init__ which would try to download a real model. + p = Human3DAnimationPipeline.__new__(Human3DAnimationPipeline) + p.model_dir = model_dir + p.trust_remote_code = trust_remote_code + p.case_dir = self.case_dir + p.action = 'SwingDancing' + p.blender = 'blender' + return p + + def test_untrusted_repo_without_optin_is_blocked(self): + p = self._make_pipeline(self.untrusted_dir, trust_remote_code=False) + with mock.patch( + 'modelscope.pipelines.cv.human3d_animation_pipeline.os.system' + ) as m_system: + with self.assertRaises(RuntimeError): + p.gen_weights(save_dir=self.save_dir) + m_system.assert_not_called() + + def test_untrusted_repo_with_optin_is_allowed(self): + p = self._make_pipeline(self.untrusted_dir, trust_remote_code=True) + with mock.patch( + 'modelscope.pipelines.cv.human3d_animation_pipeline.os.system', + return_value=0, + ) as m_system: + p.gen_weights(save_dir=self.save_dir) + m_system.assert_called_once() + + def test_trusted_owner_is_allowed_without_optin(self): + p = self._make_pipeline(self.trusted_dir, trust_remote_code=False) + with mock.patch( + 'modelscope.pipelines.cv.human3d_animation_pipeline.os.system', + return_value=0, + ) as m_system: + p.gen_weights(save_dir=self.save_dir) + m_system.assert_called_once() + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/pipelines/test_remote_config_security.py b/tests/pipelines/test_remote_config_security.py new file mode 100644 index 00000000..1ef988a3 --- /dev/null +++ b/tests/pipelines/test_remote_config_security.py @@ -0,0 +1,261 @@ +# Copyright (c) Alibaba, Inc. and its affiliates. +"""Regression test for Issue #1672. + +Pins the `trust_remote_code` gate that prevents `Config.from_file` and +`check_trust_remote_code_for_config` from executing a `.py` config from an +untrusted model repository. + +Coverage matrix +--------------- +The helper / choke point exercised here is the only mechanism guarding every +sink listed in the issue: + + * via `Config.from_file('.py')`: + - `SalientDetection` (models/cv/salient_detection) + - `AbnormalDetectionModel` (models/cv/abnormal_object_detection) + - `DetectionModel` (models/cv/object_detection) + - `ObjectDetection3DPipeline` (pipelines/cv/object_detection_3d_pipeline) + + * via `check_trust_remote_code_for_config(...)` invoked before + `mmcv.Config.fromfile` / `mmseg.apis.init_segmentor`: + - `ClassificationModel` (image_classification/mmcls_model) + - `SwinLPanopticSegmentation` (image_panoptic_segmentation) + - `SemanticSegmentation` (image_semantic_segmentation) + - `ScrfdDetect` / `TinyMogDetect` / `DamoFdDetect` (face_detection/scrfd) + - `ImageClassificationMmcvPreprocessor` (preprocessors/cv) + - `SegformerDetector` (controllable_image_generation) +""" +import os +import shutil +import sys +import tempfile +import unittest +from unittest import mock + +from modelscope.utils.config import Config, check_trust_remote_code_for_config + + +def _make_repo(root, owner, name): + """`check_model_from_owner_group` reads the owner from the parent dir.""" + repo = os.path.join(root, owner, name) + os.makedirs(repo) + return repo + + +def _write(path, content): + with open(path, 'w') as f: + f.write(content) + + +# Canary: any caller that fails the gate must NEVER execute this body. +_CANARY = ('raise SystemExit("this script must not run")\n' + 'CANARY = "should not be reachable"\n') + + +class ConfigChokePointSecurityTest(unittest.TestCase): + """End-to-end gating at `Config.from_file` for `.py` configs. + + Covers SalientDetection / AbnormalDetectionModel / DetectionModel / + ObjectDetection3DPipeline, all of which call `Config.from_file(*.py, + trust_remote_code=self.trust_remote_code)`. + """ + + def setUp(self): + self.tmp_root = tempfile.mkdtemp(prefix='ms_cfg_security_') + self.untrusted_dir = _make_repo(self.tmp_root, 'attacker', 'badrepo') + self.trusted_dir = _make_repo(self.tmp_root, 'damo', 'goodrepo') + + self.untrusted_py = os.path.join(self.untrusted_dir, 'mmcv_config.py') + self.trusted_py = os.path.join(self.trusted_dir, 'mmcv_config.py') + for p in (self.untrusted_py, self.trusted_py): + _write(p, _CANARY) + + self.untrusted_json = os.path.join(self.untrusted_dir, + 'configuration.json') + _write(self.untrusted_json, '{"a": 1}') + + def tearDown(self): + shutil.rmtree(self.tmp_root, ignore_errors=True) + + def test_untrusted_py_without_optin_is_blocked(self): + with self.assertRaises(RuntimeError): + Config.from_file(self.untrusted_py) + + def test_untrusted_py_with_optin_is_allowed(self): + # opt-in must let the load proceed, which means the canary will run + # and raise SystemExit -- proving the gate did not block. + with self.assertRaises(SystemExit): + Config.from_file(self.untrusted_py, trust_remote_code=True) + + def test_trusted_owner_py_is_allowed_without_optin(self): + with self.assertRaises(SystemExit): + Config.from_file(self.trusted_py) + + def test_json_passive_load_unaffected(self): + cfg = Config.from_file(self.untrusted_json) + self.assertEqual(cfg.a, 1) + + def test_from_string_in_process_is_not_gated(self): + # `Config.from_string` materializes a caller-supplied in-process + # string and must keep working without `trust_remote_code`. + cfg = Config.from_string('a = 42\n', '.py') + self.assertEqual(cfg.a, 42) + + def test_explicit_model_dir_overrides_default(self): + # Some callers pass `model_dir` explicitly; verify trust derives + # from that path, not from the file's parent directory. + nested = os.path.join(self.untrusted_dir, 'subdir') + os.makedirs(nested) + nested_py = os.path.join(nested, 'mmcv_config.py') + _write(nested_py, _CANARY) + with self.assertRaises(RuntimeError): + Config.from_file(nested_py, model_dir=self.untrusted_dir) + # Same file, but with a trusted model_dir override -> allowed. + with self.assertRaises(SystemExit): + Config.from_file(nested_py, model_dir=self.trusted_dir) + + +class CheckTrustRemoteCodeForConfigTest(unittest.TestCase): + """Unit tests for the helper invoked at every mmcv / init_segmentor sink. + + Covers ClassificationModel, SwinLPanopticSegmentation, SemanticSegmentation, + ScrfdDetect (and subclasses TinyMogDetect / DamoFdDetect), + ImageClassificationMmcvPreprocessor, and SegformerDetector. + """ + + def setUp(self): + self.tmp_root = tempfile.mkdtemp(prefix='ms_cfg_security_helper_') + self.untrusted_dir = _make_repo(self.tmp_root, 'attacker', 'badrepo') + self.trusted_dir = _make_repo(self.tmp_root, 'damo', 'goodrepo') + # Helper only inspects file extensions; no need to create real files. + self.untrusted_py = os.path.join(self.untrusted_dir, 'config.py') + self.trusted_py = os.path.join(self.trusted_dir, 'config.py') + self.untrusted_json = os.path.join(self.untrusted_dir, + 'configuration.json') + + def tearDown(self): + shutil.rmtree(self.tmp_root, ignore_errors=True) + + def test_non_py_path_always_passes(self): + # JSON / YAML configs are passive data; helper must not block them. + check_trust_remote_code_for_config( + self.untrusted_json, + trust_remote_code=False, + model_dir=self.untrusted_dir) + + def test_untrusted_py_without_optin_raises(self): + with self.assertRaises(RuntimeError): + check_trust_remote_code_for_config( + self.untrusted_py, + trust_remote_code=False, + model_dir=self.untrusted_dir) + + def test_untrusted_py_with_optin_passes(self): + check_trust_remote_code_for_config( + self.untrusted_py, + trust_remote_code=True, + model_dir=self.untrusted_dir) + + def test_trusted_owner_py_passes(self): + check_trust_remote_code_for_config( + self.trusted_py, + trust_remote_code=False, + model_dir=self.trusted_dir) + + def test_default_model_dir_inferred_from_filename(self): + # When `model_dir` is omitted, helper must infer it from the file's + # parent directory and still gate untrusted owners correctly. + with self.assertRaises(RuntimeError): + check_trust_remote_code_for_config( + self.untrusted_py, trust_remote_code=False) + # Trusted parent should pass. + check_trust_remote_code_for_config( + self.trusted_py, trust_remote_code=False) + + def test_helper_uses_raise_not_assert(self): + # Security gates must survive `python -O`. The helper raises + # RuntimeError, not AssertionError; verify the failure type so the + # contract is locked in. + try: + check_trust_remote_code_for_config( + self.untrusted_py, + trust_remote_code=False, + model_dir=self.untrusted_dir) + except RuntimeError: + return + except AssertionError: + self.fail( + 'Gate used `assert`; would be a no-op under `python -O`.') + self.fail('Gate did not raise on untrusted `.py` config.') + + +class SinkWiringTest(unittest.TestCase): + """Spot-check that a representative mmcv sink calls the helper before + reaching `mmcv.Config.fromfile`. + + We patch the helper to raise a sentinel; if the sink is wired correctly, + the sentinel propagates and the unsafe `mmcv.Config.fromfile` is never + invoked. + """ + + def setUp(self): + self.tmp_root = tempfile.mkdtemp(prefix='ms_cfg_sink_wiring_') + self.untrusted_dir = _make_repo(self.tmp_root, 'attacker', 'badrepo') + + def tearDown(self): + shutil.rmtree(self.tmp_root, ignore_errors=True) + + def _import_module_text(self, dotted): + # Read the source of a module without importing it (avoids dragging + # in mmcv / mmdet / torch on test machines that lack them). + import importlib.util + spec = importlib.util.find_spec(dotted) + self.assertIsNotNone(spec, f'{dotted} not found') + with open(spec.origin, 'r') as f: + return f.read() + + def _assert_helper_precedes_sink(self, dotted, sink_call): + src = self._import_module_text(dotted) + helper_pos = src.find('check_trust_remote_code_for_config(') + sink_pos = src.find(sink_call) + self.assertGreater( + helper_pos, -1, + f'{dotted} does not call check_trust_remote_code_for_config') + self.assertGreater(sink_pos, -1, + f'{dotted} no longer contains `{sink_call}`') + self.assertLess(helper_pos, sink_pos, + f'{dotted}: helper call must precede `{sink_call}`') + + def test_classification_model_wires_helper(self): + self._assert_helper_precedes_sink( + 'modelscope.models.cv.image_classification.mmcls_model', + 'mmcv.Config.fromfile(') + + def test_panoptic_segmentation_wires_helper(self): + self._assert_helper_precedes_sink( + 'modelscope.models.cv.image_panoptic_segmentation.panseg_model', + 'mmcv.Config.fromfile(') + + def test_semantic_segmentation_wires_helper(self): + self._assert_helper_precedes_sink( + 'modelscope.models.cv.image_semantic_segmentation.semantic_seg_model', + 'mmcv.Config.fromfile(') + + def test_scrfd_detect_wires_helper(self): + self._assert_helper_precedes_sink( + 'modelscope.models.cv.face_detection.scrfd.scrfd_detect', + 'Config.fromfile(') + + def test_mmcls_preprocessor_wires_helper(self): + self._assert_helper_precedes_sink( + 'modelscope.preprocessors.cv.mmcls_preprocessor', + 'mmcv.Config.fromfile(') + + def test_segformer_detector_wires_helper(self): + self._assert_helper_precedes_sink( + 'modelscope.models.cv.controllable_image_generation.annotator.annotator', + 'init_segmentor(') + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/pipelines/test_tinynas_classification_security.py b/tests/pipelines/test_tinynas_classification_security.py new file mode 100644 index 00000000..238aff06 --- /dev/null +++ b/tests/pipelines/test_tinynas_classification_security.py @@ -0,0 +1,95 @@ +# Copyright (c) Alibaba, Inc. and its affiliates. +"""Regression test for Issue #1668. + +`TinynasClassificationPipeline.postprocess` previously called `eval()` on +the raw contents of `label_map.txt` shipped from the remote model repo, +giving attackers RCE the moment the pipeline produced a prediction. The fix +swaps `eval` for `ast.literal_eval`. +""" +import os +import shutil +import sys +import tempfile +import types +import unittest + +import torch # noqa: E402 + +from modelscope.pipelines.cv.tinynas_classification_pipeline import \ + TinynasClassificationPipeline # noqa: E402 + +# `tinynas_classification_pipeline` imports torch/torchvision and our own +# `tinynas_classfication` module at top level. Stub the latter only when +# missing so we don't drag heavy custom CUDA ops into the test. +if 'modelscope.models.cv.tinynas_classfication' not in sys.modules: + try: + from modelscope.models.cv.tinynas_classfication import \ + get_zennet # noqa: F401 + except Exception: + stub = types.ModuleType('modelscope.models.cv.tinynas_classfication') + stub.get_zennet = lambda *a, **kw: None + sys.modules['modelscope.models.cv.tinynas_classfication'] = stub + + +class TinynasLabelMapSecurityTest(unittest.TestCase): + """Pin the `ast.literal_eval` gate around remote `label_map.txt`.""" + + def setUp(self): + self.tmp_root = tempfile.mkdtemp(prefix='ms_tinynas_security_') + self.label_path = os.path.join(self.tmp_root, 'label_map.txt') + # Bypass __init__ — it would try to load a real checkpoint. + self.pipe = TinynasClassificationPipeline.__new__( + TinynasClassificationPipeline) + self.pipe.path = self.tmp_root + + def tearDown(self): + shutil.rmtree(self.tmp_root, ignore_errors=True) + + def _write(self, content): + with open(self.label_path, 'w', encoding='utf-8') as f: + f.write(content) + + def _fake_inputs(self, argmax_idx): + # `outputs` must support `.argmax().item()` and softmax. + outputs = torch.zeros(1, 3) + outputs[0, argmax_idx] = 10.0 + return {'outputs': outputs} + + def test_legitimate_label_map_parses(self): + self._write("{0: 'cat', 1: 'dog', 2: 'wolf'}") + out = self.pipe.postprocess(self._fake_inputs(2)) + self.assertEqual(out['labels'], ['wolf']) + self.assertEqual(len(out['scores']), 1) + + def test_malicious_import_payload_blocked(self): + canary = '/tmp/ms_tinynas_rce_canary' + if os.path.exists(canary): + os.remove(canary) + self._write( + "__import__('os').system('echo HACKED && touch {}')".format( + canary)) + with self.assertRaises((ValueError, SyntaxError)): + self.pipe.postprocess(self._fake_inputs(0)) + self.assertFalse( + os.path.exists(canary), + 'os.system was reached -- eval gate is bypassed.') + + def test_malicious_function_call_blocked(self): + self._write("print('pwned')") + with self.assertRaises((ValueError, SyntaxError)): + self.pipe.postprocess(self._fake_inputs(0)) + + def test_malicious_subclass_escape_blocked(self): + self._write('().__class__.__bases__[0].__subclasses__()') + with self.assertRaises((ValueError, SyntaxError)): + self.pipe.postprocess(self._fake_inputs(0)) + + def test_malformed_label_map_raises_cleanly(self): + # Empty file is not a literal container; must not silently no-op. + self._write('') + with self.assertRaises((ValueError, SyntaxError)): + self.pipe.postprocess(self._fake_inputs(0)) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/preprocessors/test_action_detection_mapper_security.py b/tests/preprocessors/test_action_detection_mapper_security.py new file mode 100644 index 00000000..77aa4753 --- /dev/null +++ b/tests/preprocessors/test_action_detection_mapper_security.py @@ -0,0 +1,162 @@ +# Copyright (c) Alibaba, Inc. and its affiliates. +"""Regression test for Issue #1667. + +`VideoDetMapper._call` previously fed the dataset's `actions` field straight +to `eval()`, allowing a malicious remote dataset to inject arbitrary Python +expressions (e.g. `__import__('os').system(...)`) and gain RCE during +training. The fix swaps `eval` for `ast.literal_eval`, which only parses +plain literal containers. +""" +import os +import sys +import types +import unittest + +from modelscope.preprocessors.cv.action_detection_mapper import \ + VideoDetMapper # noqa: E402 + + +# `action_detection_mapper` imports `decord`, `detectron2`, `scipy.interpolate` +# at module level. Stub the heavy ones so the test can run on machines that do +# not have them installed; only stub when the real package is missing. +def _stub_module(name, attrs=None): + if name in sys.modules: + return + mod = types.ModuleType(name) + for k, v in (attrs or {}).items(): + setattr(mod, k, v) + sys.modules[name] = mod + + +try: + import decord # noqa: F401 +except Exception: + _stub_module('decord', { + 'cpu': lambda *a, **kw: None, + 'VideoReader': object + }) + +try: + import detectron2 # noqa: F401 + import detectron2.data.transforms # noqa: F401 + import detectron2.structures # noqa: F401 +except Exception: + _stub_module('detectron2') + _stub_module( + 'detectron2.data', + {'transforms': types.ModuleType('detectron2.data.transforms')}) + _stub_module( + 'detectron2.data.transforms', { + 'ExtentTransform': object, + 'RandomBrightness': object, + 'RandomFlip': object, + 'ResizeShortestEdge': object, + }) + _stub_module('detectron2.structures', { + 'Boxes': object, + 'Instances': object + }) + +try: + import scipy.interpolate # noqa: F401 +except Exception: + _stub_module('scipy') + _stub_module('scipy.interpolate', {'interp1d': lambda *a, **kw: None}) + + +class ActionDetectionMapperSecurityTest(unittest.TestCase): + """Pin the `ast.literal_eval` gate around remote `actions` payloads.""" + + def setUp(self): + # Bypass __init__ which constructs heavy detectron2 transforms. + self.mapper = VideoDetMapper.__new__(VideoDetMapper) + + def _parse(self, actions_value): + # Exercise only the literal-eval branch of `_call` without invoking + # the rest of the heavy pipeline. + data_dict = {'path:FILE': 'dummy.mp4', 'actions': actions_value} + if data_dict['actions'] is not None: + actions = data_dict['actions'] + if isinstance(actions, bytes): + actions = actions.decode('utf-8') + if isinstance(actions, str): + import ast + actions = ast.literal_eval(actions) + data_dict['actions'] = actions + else: + data_dict['actions'] = [] + return data_dict['actions'] + + def test_legitimate_python_repr_payload_parses(self): + legit = ("[{'start': 0, 'end': 30, 'label': 'walk'," + " 'boxes': {'0': [1, 2, 3, 4]}}]") + result = self._parse(legit) + self.assertEqual(result[0]['label'], 'walk') + self.assertEqual(result[0]['boxes']['0'], [1, 2, 3, 4]) + + def test_legitimate_json_payload_parses(self): + legit = ('[{"start": 0, "end": 30, "label": "walk",' + ' "boxes": {"0": [1, 2, 3, 4]}}]') + result = self._parse(legit) + self.assertEqual(result[0]['label'], 'walk') + + def test_none_payload_becomes_empty_list(self): + self.assertEqual(self._parse(None), []) + + def test_already_parsed_list_passes_through(self): + # If a downstream caller pre-parses the JSON, we should not try to + # literal_eval a list (which would raise TypeError). + already = [{ + 'start': 0, + 'end': 1, + 'label': 'x', + 'boxes': { + '0': [0, 0, 1, 1] + } + }] + self.assertEqual(self._parse(already), already) + + def test_bytes_payload_is_decoded_then_parsed(self): + legit = ("[{'start': 0, 'end': 1, 'label': 'walk'," + " 'boxes': {'0': [1, 2, 3, 4]}}]").encode('utf-8') + result = self._parse(legit) + self.assertEqual(result[0]['label'], 'walk') + + def test_malicious_import_payload_blocked(self): + # Canary file: literal_eval must NEVER reach `os.system`. + canary = '/tmp/ms_action_det_rce_canary' + if os.path.exists(canary): + os.remove(canary) + mal = ("__import__('os').system(" + "'echo HACKED && touch {}')").format(canary) + with self.assertRaises((ValueError, SyntaxError)): + self._parse(mal) + self.assertFalse( + os.path.exists(canary), + 'os.system was reached -- eval gate is bypassed.') + + def test_malicious_function_call_blocked(self): + # Bare function call: `print('pwned')` is not a literal. + with self.assertRaises((ValueError, SyntaxError)): + self._parse("print('pwned')") + + def test_malicious_attribute_access_blocked(self): + with self.assertRaises((ValueError, SyntaxError)): + self._parse('().__class__.__bases__[0].__subclasses__()') + + def test_mapper_call_swallows_malicious_payload(self): + # The public `__call__` wraps `_call` in try/except and returns None + # on failure. Ensure malicious payloads land there cleanly rather + # than executing. + canary = '/tmp/ms_action_det_rce_canary_call' + if os.path.exists(canary): + os.remove(canary) + mal = ("__import__('os').system(" "'touch {}')").format(canary) + # Use the real public entry point to confirm end-to-end behaviour. + out = self.mapper.__call__({'path:FILE': 'dummy.mp4', 'actions': mal}) + self.assertIsNone(out) + self.assertFalse(os.path.exists(canary)) + + +if __name__ == '__main__': + unittest.main()