Vulnerable code sample

import os
import sys

# This simulates the logic for defining the root path for internal assets.
if getattr(sys, 'frozen', False):
    EMMETT_ASSETS_PATH = os.path.join(sys._MEIPASS, 'assets')
else:
    EMMETT_ASSETS_PATH = os.path.join(os.path.dirname(__file__), 'assets')


async def vulnerable_rsgi_static_handler(scope, receive, send):
    """
    This function is a representation of the vulnerable RSGI static file handler
    in Emmett versions prior to 2.8.1. It is intended for demonstration purposes only.
    """
    if scope['type'] != 'http':
        return

    path = scope['path']

    if not path.startswith('/__emmett__/'):
        return

    relative_path = path[len('/__emmett__/'):]

    # The vulnerability lies here: user input is joined to a path without sanitization.
    # An attacker can use '..' to traverse the filesystem.
    file_path = os.path.join(EMMETT_ASSETS_PATH, relative_path)

    # To avoid 'file_path' being an absolute path, which os.path.join allows
    # if the second argument starts with '/', we naively strip it.
    # This was part of the original logic that still allowed traversal.
    if os.path.isabs(file_path):
        file_path = file_path.lstrip('/')

    try:
        with open(file_path, "rb") as file:
            content = file.read()

        await send({
            'type': 'http.response.start',
            'status': 200,
            'headers': [
                (b'content-type', b'application/octet-stream')
            ]
        })
        await send({'type': 'http.response.body', 'body': content})

    except (FileNotFoundError, IsADirectoryError):
        await send({
            'type': 'http.response.start',
            'status': 404,
            'headers': [(b'content-type', b'text/plain')]
        })
        await send({'type': 'http.response.body', 'body': b'Not Found'})