DDCTF2019 两道WEB题解
DDCTF2019 两道WEB题解 前几天打了DDCTF,有几道WEB题还是挺不错的,在这里分析一下。 homebrew event loop 题目直接给了源码,是一道flask代码审计 # -*- encoding: utf-8 -*- # written in python 2.7 __author__ = 'garzon' from flask import Flask, sessi
DDCTF2019 两道WEB题解
前几天打了DDCTF,有几道WEB题还是挺不错的,在这里分析一下。
homebrew event loop
题目直接给了源码,是一道flask代码审计
# -*- encoding: utf-8 -*-
# written in python 2.7
__author__ = 'garzon'
from flask import Flask, session, request, Response
import urllib
app = Flask(__name__)
app.secret_key = '*********************' # censored
url_prefix = '/d5af31f88147e857'
def FLAG():
return 'FLAG_is_here_but_i_wont_show_you' # censored
def trigger_event(event):
session['log'].append(event)
if len(session['log']) > 5: session['log'] = session['log'][-5:]
if type(event) == type([]):
request.event_queue += event
else:
request.event_queue.append(event)
def get_mid_str(haystack, prefix, postfix=None):
haystack = haystack[haystack.find(prefix)+len(prefix):]
if postfix is not None:
haystack = haystack[:haystack.find(postfix)]
return haystack
class RollBackException: pass
def execute_event_loop():
valid_event_chars = set('abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ_0123456789:;#')
resp = None
while len(request.event_queue) > 0:
event = request.event_queue[0] # `event` is something like "action:ACTION;ARGS0#ARGS1#ARGS2......"
request.event_queue = request.event_queue[1:]
if not event.startswith(('action:', 'func:')): continue
for c in event:
if c not in valid_event_chars: break
else:
is_action = event[0] == 'a'
action = get_mid_str(event, ':', ';')
args = get_mid_str(event, action+';').split('#')
try:
event_handler = eval(action + ('_handler' if is_action else '_function'))
ret_val = event_handler(args)
except RollBackException:
if resp is None: resp = ''
resp += 'ERROR! All transactions have been cancelled. '
resp += '<a href="./?action:view;index">G'
session['num_items'] = request.prev_session['num_items']
session['points'] = request.prev_session['points']
break
except Exception, e:
if resp is None: resp = ''
#resp += str(e) # only for debugging
continue
if ret_val is not None:
if resp is None: resp = ret_val
else: resp += ret_val
if resp is None or resp == '': resp = ('404 NOT FOUND', 404)
session.modified = True
return resp
@app.route(url_prefix+'/')
def entry_point():
querystring = urllib.unquote(request.query_string)
request.event_queue = []
if querystring == '' or (not querystring.startswith('action:')) or len(querystring) > 100:
querystring = 'action:index;False#False'
if 'num_items' not in session:
session['num_items'] = 0
session['points'] = 3
session['log'] = []
request.prev_session = dict(session)
trigger_event(querystring)
return execute_event_loop()
# handlers/functions below --------------------------------------
def view_handler(args):
page = args[0]
html = ''
html += '[INFO] you have {} diamonds, {} points now.'.format(session['num_items'], session['points'])
if page == 'index':
html += '<a href="./?action:index;True%23Fal'
html += '<a href="./?action:vie'
html += '<a href="./?acti'
elif page == 'shop':
html += '<a href="./?action:buy;1">Buy'
elif page == 'reset':
del session['num_items']
html += 'Session reset.'
html += '<a href="./?action:view;index">G'
return html
def index_handler(args):
bool_show_source = str(args[0])
bool_download_source = str(args[1])
if bool_show_source == 'True':
source = open('eventLoop.py', 'r')
html = ''
if bool_download_source != 'True':
html += '<a href="./?action:index;True%23True">Do'
html += '<a href="./?action:view;index">G'
for line in source:
if bool_download_source != 'True':
html += line.replace('&','&').replace('\t', ' '*4).replace(' ',' ').replace('<', '<').replace('>','>').replace('\n', '')
else:
html += line
source.close()
if bool_download_source == 'True':
headers = {}
headers['Content-Type'] = 'text/plain'
headers['Content-Disposition'] = 'attachment; filename=serve.py'
return Response(html, headers=headers)
else:
return html
else:
trigger_event('action:view;index')
def buy_handler(args):
num_items = int(args[0])
if num_items <= 0: return 'invalid number({}) of diamonds to buy'.format(args[0])
session['num_items'] += num_items
trigger_event(['func:consume_point;{}'.format(num_items), 'action:view;index'])
def consume_point_function(args):
point_to_consume = int(args[0])
if session['points'] < point_to_consume: raise RollBackException()
session['points'] -= point_to_consume
def show_flag_function(args):
flag = args[0]
#return flag # GOTCHA! We noticed that here is a backdoor planted by a hacker which will print the flag, so we disabled it.
return 'You naughty boy! ;) '
def get_flag_handler(args):
if session['num_items'] >= 5:
trigger_event('func:show_flag;' + FLAG()) # show_flag_function has been disabled, no worries
trigger_event('action:view;index')
if __name__ == '__main__':
app.run(debug=False, host='0.0.0.0')
FLAG()函数会返回flag,但是需要想办法执行他,并获取返回值。trigger_event函数会把收到的参数存入session['log'],然后存入队列中。
并且源码中只有一个路由url_prefix+'/',url参数需要以action:开头,并且url参数会直接全部传入trigger_event中,最终会返回execute_event_loop()函数。
可以看到这个函数会循环提取队列中的字符串,最终由get_mid_str函数提取出函数名和参数,然后把函数名用eval与_handler或者_function拼接,接着执行该函数。
看一下get_flag_handler函数,当session['num_items'] >= 5会把flag传入trigger_event,然后会存入session,我们把session解码即可看到flag。
这里有比较关键的两个函数buy_handler和consume_point_function,我们的points初始为3,session['num_items']为0,每一次buy的参数要小于points的值,否则会报错。
现在我们的思路是:要么直接执行FLAG()函数把flag返回到前端,要么在buy_handler一个很大的参数之后直接调用get_flag_handler。
直接执行FLAG()函数
从上面到测试中可以看到,在eval中#号会注释掉后面掉字符串,也就是绕过函数名字符串拼接,直接执行任意函数。但是我们会发现split始终返回一个列表,然后被当作函数到参数
我们发现即空列表作为参数,也无法执行该函数。
所以此路不通
buy_handler->get_flag_handler
我们知道我们到url参数会被直接传入队列,并且现在我们可以调用任意函数。
看一下get_mid_str的实现
会直接返回第一个;之后的内容,接着用#号分割为列表。而我们的trigger_event是支持传入列表的,那么我们可以调用名为trigger_event的函数,参数为先buy后get_flag即可。
payload:?action:trigger_event%23;action:buy;5%23action:get_flag;,访问之后session解码即可。
mysql弱口令
这道题用到的是MySQL LOAD DATA 读取客户端任意文件需要注意的是agent.py中的Process_name需要含有mysqld,直接改源码,端口写3306,然后跑https://github.com/allyshka/Rogue-MySql-Server中的脚本即可。
接下来就是找flag,可以直接读~/.mysql_history
或者读取~/.bash_history,找到工作目录,读源码
/home/dc2-user/ctf_web_2/app/main/views.py
# coding=utf-8
from flask import jsonify, request
from struct import unpack
from socket import inet_aton
import MySQLdb
from subprocess import Popen, PIPE
import re
import os
import base64
# flag in mysql curl@localhost database:security table:flag
def weak_scan():
agent_port = 8123
result = []
target_ip = request.args.get(\'target_ip\')
target_port = request.args.get(\'target_port\')
.......
可以看到flag在security库flag表中。my.cnf
/var/lib/mysql/security/flag.ibd
最后一道mysql讲的很不错,好像还可以在这题读吃鸡的源码,tql