ArtificialUniversity
0x00 开始之前
htb web题做完之后就很久没上htb了,最近一上发现又上新题了,还是INSANE难度的,直接下载附件开始审计。
0x01 梳理结构
首先打开项目,先大致看一下目录结构和代码,可以知道其是一个flask的web程序
然后看它的路由有
/
/login
/register
/logout
/product/<product_id>
/subs
/checkout
/checkout/success
/admin
/admin/users
/admin/products
/admin/orders
/admin/view-pdf
/admin/api-health
/admin/product-stream
/admin/save-product
/admin/saved-products
其中/admin
开头的接口都做了权限校验
到这可以知道这是一个购物相关的网站。根据路由并配合我们访问页面查看请求以及对应代码可以得到以下业务逻辑
用户侧的逻辑为:
-
用户注册:
/register
-
用户登录:
/login
-
产品商品:
/product/<product_id>
-
用户下单:
/checkout
-
用户付款:
/checkout/success
(回调) -
用户订单:
/subs
-
用户登出:
/logout
管理员侧逻辑为
-
查看用户列表:
/admin/users
-
查看所有商品:
/admin/products
-
查看所有订单:
/admin/orders
-
查看pdf:
/admin/view-pdf
-
查看api接口状态:
/admin/api-health
-
获取新产品:
/admin/product-stream
-
保存产品:
/admin/save-product
-
所有保存的产品:
/admin/saved-products
0x02 代码审计
搞清楚了业务逻辑,我们就可以开始进行代码审计工作了。
越权失败
首先,因为/admin
前缀的接口都需要admin权限校验,正常我们无法访问,我们第一个想到了就是如何获取权限或者越权访问到这些接口, 而从config.py
中的配置来看,管理员的密码是无法爆破的,并且/register
接口内注册的时候权限也是写死的为user
,无法改为admin
,其它代码也有没有修改权限相关的,那么这一条路应该不行。
# src/store/application/config.py
class Config(object):
SECRET_KEY = os.urandom(50).hex()
ADMIN_EMAIL = "[email protected]"
ADMIN_PASS = os.urandom(32).hex()
@web.route("/register", methods=["GET", "POST"])
def register():
...
user_valid = db_session.create_user(email, password)
...
def create_user(self, email, password, role="user"):
...
new_user = Users(email=email, password=password_hash, role=role)
...
1. 逻辑漏洞 + 路径穿越
越权这条路走不通,我们只能看看其它代码,经过进一步审计发现了一处逻辑漏洞。
首先,通过阅读其代码,我们可以知道,当我们不传入product_id参数的时候,它的price居然是参数是可控的,也就是说下单的时候我们可以填写任意价格,甚至是负的!
@web.route("/checkout", methods=["GET"])
def checkout():
product_id = request.args.get("product_id")
if product_id and not session.get("loggedin"):
return render_template("error.html", title="Error", error="Must have an account in order to purchase"), 200
price = request.args.get("price") # 价格怎么能可控的?!
title = request.args.get("title")
user_id = request.args.get("user_id")
email = request.args.get("email")
if not product_id and (not price or not title or not user_id or not email):
return render_template("error.html", title="Error", error="Missing external order details"), 400
db_session = Database()
payment_link = None
if product_id:
product_data = db_session.get_product_data(product_id)
if not product_data:
return render_template("error.html", title="Error", error="Product not found"), 404
payment_link, payment_id = generate_payment_link(product_data.price)
order_id = db_session.create_order(product_data.title, session.get("user_id"), session.get("email"), product_data.price, payment_id, product_data.id)
db_session.generate_invoice(order_id)
else:
product_data = {
"title": title,
"price": int(price)
}
payment_link, payment_id = generate_payment_link(product_data["price"])
order_id = db_session.create_order(title, user_id, email, int(price), payment_id)
db_session.generate_invoice(order_id)
return redirect(payment_link)
再结合下单成功的回调接口/checkout/success
,下单成功后会触发一个bot_runner,其会去访问/checkout
生成的pdf订单,而这里的payment_id也是可控并且是任意的。因为这个bot模拟的是浏览器操作,而浏览器的url也是支持目录跳转的,所以当我们传入../../../../../../admin#
的时候,浏览器解析的url就会变为http://127.0.0.1:1337/admin#pdf
,其中#号作为资源定位符也是不会传给后端的,所以后端最终接收到的就是http://127.0.0.1:1337/admin
。通过这种路径穿越,并且在访问之前,bot还贴心的帮我们进行了admin账号的登录,那我们就可以访问当前服务的任意路由了,包括admin。
def get_amount_paid(payment_id):
# Dummy implementation to get payment status
return 0
def bot_runner(email, password, payment_id):
...
firefox_service = Service(geckodriver_path)
client = webdriver.Firefox(service=firefox_service, options=firefox_options)
try:
client.get("http://127.0.0.1:1337/login")
time.sleep(3)
client.find_element(By.ID, "email").send_keys(email)
client.find_element(By.ID, "password").send_keys(password)
client.execute_script("document.getElementById('login-btn').click()")
time.sleep(3)
client.get(f"http://127.0.0.1:1337/static/invoices/invoice_{payment_id}.pdf")
time.sleep(10)
finally:
client.quit()
@web.route("/checkout/success", methods=["GET"])
def checkout_success():
order_id = request.args.get("order_id")
payment_id = request.args.get("payment_id")
if not order_id:
return render_template("error.html", title="Error", error="Missing parameters"), 401
db_session = Database()
order = db_session.get_order(order_id)
# 因为无法真正的付款,这个函数始终返回0
amt_paid = get_amount_paid(payment_id)
if amt_paid >= order.price:
db_session.mark_order_complete(order_id)
else:
return render_template("error.html", title="Error", error="Could not complete order"), 401
bot_runner(current_app.config["ADMIN_EMAIL"], current_app.config["ADMIN_PASS"], payment_id)
return render_template("success.html", title="Order successful", nav=True)
将两者结合起来我们可以实现以下逻辑:
-
首先
/checkout
接口我们不传入product_id,并且price为-1,那么他就会创建一个价格为-1的订单 -
然后我们再用生成的order_id,并将payment_id设为
../../../../../admin/xxx#
就可以访问admin相关的路由了
2. ssrf
能否访问admin接口了,我们可以继续对admin相关代码进行审计了。其中有两个接口很符合我们的当前的利用,这两个接口都是潜在的ssrf的点,其中第一个只需要get请求传入url即可,更加符合我们现在的条件(浏览器输入网址默认请求为get)
@web.route("/admin/view-pdf", methods=["GET"])
def admin_view_pdf():
if not session.get("loggedin") or session.get("role") != "admin":
return redirect("/")
pdf_url = request.args.get("url")
if not pdf_url:
return render_template("error.html", title="Error", error="Missing PDF URL"), 400
try:
response = requests.get(pdf_url)
response.raise_for_status()
if response.headers["Content-Type"] != "application/pdf":
return render_template("error.html", title="Error", error="URL does not point to a PDF file"), 400
pdf_data = BytesIO(response.content)
return send_file(pdf_data, mimetype="application/pdf", as_attachment=False, download_name="document.pdf")
except requests.RequestException as e:
return render_template("error.html", title="Error", error=str(e)), 400
@web.route("/admin/api-health", methods=["GET", "POST"])
def api_health():
if not session.get("loggedin") or session.get("role") != "admin":
return redirect("/")
if request.method == "GET":
return render_template("admin_api_health.html", title="Admin panel - API health", session=session)
url = request.form.get("url")
if not url:
return render_template("error.html", title="Error", error="Missing URL"), 400
status_code = get_url_status_code(url)
return render_template("admin_api_health.html", title="Admin panel - API health", session=session, status_code=status_code)
对于/admin/view-pdf
而言,其逻辑为传入一个url,使用requests进行访问,获取到内容后以application/pdf
的形式返回。 这就是一个ssrf的点了,但是即没有回显,又只能get,好像没什么意义。这里还有一个思路就是使用构造一个恶意的pdf,在pdf里面执行js,利用js来进行下一步利用。
但是经过不断搜索和尝试发现,我使用adobe的js api创建的js只能执行弹窗,其它啥也干不了。
最后搜索到一个pdf.js的CVE-2024-4367,在用pdf.js解析pdf的时候,通过该漏洞可以实现任意js代码执行。这个漏洞在firefox126版本才被修复,然后一看题目的dockerfile,果然安装的是存在漏洞的版本。
# Install a specific version of Firefox
ARG FIREFOX_VERSION=125.0.1
到这一步,通过公开的poc,我们就可以创建一个在bot上执行任意js的pdf了。
3. ssrf 2
bot + xss,能做什么呢,带着这个利用,我们继续深入代码审计,还记得前面两个疑似ssrf的接口,第二个接口使用的是表单post,那么这个时候我们就可以利用js来自动提交表单来请求这个接口了。(后记:其实这一步也卡了很久,主要是调试的时候使用的是浏览器的有头模式,表单自动提交会弹窗确认导致请求失败,后面换成无头的才测试成功了。)
那么这个/admin/api-health
接口的具体逻辑是什么呢,它接受一个url,并使用curl进行访问并获取其响应码,这里即不存在命令注入,也无法实现参数注入(url的特殊字符会被转义),所以也是一个ssrf,但是这个ssrf比前面的更加自由了,不仅可以GET,还可以使用其它请求,甚至原生TCP请求(万能的神gopher),给了这个利用应该是想让我们打内网吧。
@web.route("/admin/api-health", methods=["GET", "POST"])
def api_health():
if not session.get("loggedin") or session.get("role") != "admin":
return redirect("/")
if request.method == "GET":
return render_template("admin_api_health.html", title="Admin panel - API health", session=session)
url = request.form.get("url")
if not url:
return render_template("error.html", title="Error", error="Missing URL"), 400
status_code = get_url_status_code(url)
return render_template("admin_api_health.html", title="Admin panel - API health", session=session, status_code=status_code)
def is_valid_url(url):
try:
result = urlparse(url)
return all([result.scheme, result.netloc])
except ValueError:
return False
def get_url_status_code(url):
if not is_valid_url(url):
raise ValueError("Invalid URL")
curl_args = ["curl", "-o", "/dev/null", "-w", "%{http_code}", url]
try:
result = subprocess.run(curl_args, capture_output=True, text=True, check=True)
status_code = int(result.stdout.strip())
return status_code
except subprocess.CalledProcessError as e:
return "Error getting status code"
4. grpc
带着打内网的思路,我们进行看代码和dockerfile,发现,其内部还部署了一个grpc服务,第一眼就看到了这个GenerateProduct里的eval
,一看就不是啥好东西,仔细看其逻辑可以知道,当这个service存在price_formula属性的时候,会使用eval(self.price_formula)计算价格,而刚好UpdateService里就可以给一个对象添加某个属性,又刚好DebugService作为对外的服务又调用了UpdateService。然后我们再调用GetNewProducts服务,让其调用GenerateProduct函数来触发eval,这条任意代码执行的利用链就串起来了。
那么结合前面的ssrf2,我们是不是可以利用curl+gopher协议,来请求这个内网的grpc服务呢。
说干就干,因为grpc是基于http2的,只需要实现gophers发送http2流即可。
首先我们把这个grpc服务起起来,然后通过grpc客户端发送我们想要的请求,通过wireshark
,我们可以看到HTTP2的数据流,导出发送端的所有数据,转换成gopher协议就可以了。
class ProductService(product_pb2_grpc.ProductServiceServicer):
def __init__(self):
self.saved_products = []
self.max_products = 3
self.min_price = 10
self.max_price = 100
def UpdateService(self, source, destination):
for key, value in source.items():
if hasattr(destination, "__dict__") and key in destination.__dict__ and isinstance(value, dict):
self.UpdateService(value, destination.__dict__[key])
elif hasattr(destination, "__dict__"):
destination.__dict__[key] = value
elif isinstance(destination, dict) and key in destination and isinstance(value, dict):
self.UpdateService(value, destination[key])
else:
destination[key] = value
def GenerateProduct(self):
if hasattr(self, "price_formula"):
price = eval(self.price_formula) # eval?,太明显了吧
product = product_pb2.Product(
id=str(uuid.uuid4()),
name=f"Product {random.randint(1, 100)}",
description="A sample product",
price=price
)
return product
else:
product = product_pb2.Product(
id=str(uuid.uuid4()),
name=f"Product {random.randint(1, 100)}",
description="A sample product",
price=random.uniform(self.min_price, self.max_price)
)
return product
def GetNewProducts(self, request, context):
new_products = []
for i in range(random.randint(0, 3)):
new_products.append(self.GenerateProduct())
return product_pb2.Products(products=new_products)
def MarkProductSaved(self, request, context):
if len(self.saved_products) >= self.max_products:
self.saved_products.pop(0)
self.saved_products.append(request)
return product_pb2.Empty()
def GetSavedProducts(self, request, context):
return product_pb2.Products(products=self.saved_products)
def DebugService(self, request, context):
input_dict = {k: v.string_value for k, v in request.input.items()}
self.UpdateService(input_dict, self)
return product_pb2.Empty()
def serve():
server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
product_pb2_grpc.add_ProductServiceServicer_to_server(ProductService(), server)
server.add_insecure_port("[::]:50051")
server.start()
server.wait_for_termination()
if __name__ == "__main__":
serve()
5. 利用链
终于,我们由外到内,串联一个又一个脆弱点,最终实现了任意代码执行。总结一下所有的漏洞:
-
逻辑漏洞导致的负数订单创建
-
支付回调接口缺少校验导致的ssrf(限制host)
-
使用漏洞组件导致的xss
-
内网服务使用不安全的函数eval
0x03 总结
只能说不愧是INSANE难度的题目,一环扣一环,一个点接着一个点,让人又折磨又享受,中间还有也很多遇到的坑没有写到。exp我就不贴了,推荐各位ctf爱好者上手审一审。
0x04 参考链接
CVE-2024-4367 – Arbitrary JavaScript execution in PDF.js
https://www.mozilla.org/en-US/security/advisories/mfsa2024-21/#CVE-2024-4367
4A评测 - 免责申明
本站提供的一切软件、教程和内容信息仅限用于学习和研究目的。
不得将上述内容用于商业或者非法用途,否则一切后果请用户自负。
本站信息来自网络,版权争议与本站无关。您必须在下载后的24个小时之内,从您的电脑或手机中彻底删除上述内容。
如果您喜欢该程序,请支持正版,购买注册,得到更好的正版服务。如有侵权请邮件与我们联系处理。敬请谅解!
程序来源网络,不确保不包含木马病毒等危险内容,请在确保安全的情况下或使用虚拟机使用。
侵权违规投诉邮箱:4ablog168#gmail.com(#换成@)