Fix some security issues (#1737)

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]:

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'))

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):

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)

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)

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

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

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)
 

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

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

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)

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:

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)

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'] = []
 

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]:

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'

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
 

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()

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()

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()

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()