密码学

1.AES

#!/usr/bin/env python3
from secret import flag
from Crypto.Cipher import AES
from Crypto.Random import get_random_bytes

FIXED_KEY = get_random_bytes(16)  
FIXED_IV = get_random_bytes(16) 

class VulnerableGCMCipher:

    def __init__(self):
        self.key = FIXED_KEY
        self.iv = FIXED_IV 
        
    def encrypt(self, plaintext):
        cipher = AES.new(self.key, AES.MODE_GCM, nonce=self.iv)
        ciphertext = cipher.encrypt_and_digest(plaintext.encode() if isinstance(plaintext, str) else plaintext)
        return ciphertext

def generate_data():
    cipher_service = VulnerableGCMCipher()

    known_message = "The flag is hidden somewhere in this encrypted system."
    known_ciphertext = cipher_service.encrypt(known_message)

    flag_ciphertext = cipher_service.encrypt(flag)
    
    print(f"已知密文: {known_ciphertext.hex()}")
    print(f"目标密文: {flag_ciphertext.hex()}")

if __name__ == "__main__":
    data = generate_data()

'''
已知密文: b7eb5c9e8ea16f3dec89b6dfb65670343efe2ea88e0e88c490da73287c86e8ebf375ea1194b0d8b14f8b6329a44f396683f22cf8adf8
目标密文: 85ef58d9938a4d1793a993a0ac0c612368cf3fa8be07d9dd9f8c737d299cd9adb76fdc1187b6c3a00c866a20
'''

def xor_bytes(a, b):
    return bytes(x ^ y for x, y in zip(a, b))

known_ciphertext = bytes.fromhex("b7eb5c9e8ea16f3dec89b6dfb65670343efe2ea88e0e88c490da73287c86e8ebf375ea1194b0d8b14f8b6329a44f396683f22cf8adf8")

known_plaintext = "The flag is hidden somewhere in this encrypted system."
known_plaintext_bytes = known_plaintext.encode()

keystream = xor_bytes(known_plaintext_bytes, known_ciphertext)

print("(keystream):")
print(keystream.hex())

target_ciphertext = bytes.fromhex("85ef58d9938a4d1793a993a0ac0c612368cf3fa8be07d9dd9f8c737d299cd9adb76fdc1187b6c3a00c866a20")
keystream_extended = (keystream * ((len(target_ciphertext) // len(keystream)) + 1))[:len(target_ciphertext)]

decrypted_bytes = xor_bytes(target_ciphertext, keystream_extended)
try:
    decrypted_message = decrypted_bytes.decode()
    print(decrypted_message)
except UnicodeDecodeError:
    print(decrypted_bytes)

(keystream):

e38339bee8cd0e5acce0c5ffde3f14505b900edbe163edb3f8bf014d5cef86cb871d8362b4d5b6d23df2135dc12b1915fa81589dc0d6

flag{GCM_IV_r3us3_1s_d4ng3r0us_f0r_s3cur1ty}

2.rsadl

from Crypto.Util.number import *
import gmpy2
from secrets import flag

p = getPrime(512)
q = getPrime(512)
e = 65537

n = p*q
m = bytes_to_long(flag)
c = gmpy2.powmod(m,e,n)

phin = (p - 1) * (q - 1)
d = gmpy2.invert(e, phin)


print("n = ", n)
print("c = ", c)
print("dl =  ", d & ((1 << 530) - 1))

# n = 143504495074135116523479572513193257538457891976052298438652079929596651523432364937341930982173023552175436173885654930971376970322922498317976493562072926136659852344920009858340197366796444840464302446464493305526983923226244799894266646253468068881999233902997176323684443197642773123213917372573050601477
# c = 141699518880360825234198786612952695897842876092920232629929387949988050288276438446103693342179727296549008517932766734449401585097483656759727472217476111942285691988125304733806468920104615795505322633807031565453083413471250166739315942515829249512300243607424590170257225854237018813544527796454663165076
# dl = 1761714636451980705225596515441824697034096304822566643697981898035887055658807020442662924585355268098963915429014997296853529408546333631721472245329506038801

import math
from Cryptodome.Util.number import long_to_bytes

def is_perfect_square(num):
    """判断num是否为完全平方数，返回(bool, sqrt(num))"""
    if num < 0:
        return False, 0
    root = math.isqrt(num)
    return root * root == num, root

def rsa_low_d_partial_recover(n, e, bits_leaked, leaked_d_low, ciphertext=None):
    base = e * (1 << bits_leaked)
    residue = (e * leaked_d_low - 1) % base
    sqrt_n = math.isqrt(n)

    for k in range(1, e):
        g = math.gcd(k, base)
        if residue % g != 0:
            continue
        base1, k1, residue1 = base // g, k // g, residue // g
        if math.gcd(k1, base1) != 1:
            continue
        inv_k1 = pow(k1, -1, base1)
        phi_partial = (residue1 * inv_k1) % base1
        z_estimate = (n - phi_partial) // base1
        for dz in range(-3, 4):
            phi = phi_partial + base1 * (z_estimate + dz)
            if not (1 < phi < n):
                continue
            delta = (n - phi + 1)
            is_sq, sqrtval = is_perfect_square(delta * delta - 4 * n)
            if not is_sq:
                continue
            p = (delta - sqrtval) // 2
            q = (delta + sqrtval) // 2
            if p * q != n:
                continue
            priv_d = pow(e, -1, (p - 1) * (q - 1))
            if ciphertext is not None:
                m = pow(ciphertext, priv_d, n)
                return long_to_bytes(m)
            else:
                return priv_d, (p, q)
    return None

# 参数定义
n = 143504495074135116523479572513193257538457891976052298438652079929596651523432364937341930982173023552175436173885654930971376970322922498317976493562072926136659852344920009858340197366796444840464302446464493305526983923226244799894266646253468068881999233902997176323684443197642773123213917372573050601477
c = 141699518880360825234198786612952695897842876092920232629929387949988050288276438446103693342179727296549008517932766734449401585097483656759727472217476111942285691988125304733806468920104615795505322633807031565453083413471250166739315942515829249512300243607424590170257225854237018813544527796454663165076
e = 65537
leak_bits = 530
d_low = 1761714636451980705225596515441824697034096304822566643697981898035887055658807020442662924585355268098963915429014997296853529408546333631721472245329506038801

# 调用解密
print(rsa_low_d_partial_recover(n, e, leak_bits, d_low, c))

3.多重Caesar密码

“myfz”对应“flag”，用的是依次前移7、13、5、19位，“pfxddi”对应“caesar”用的是13、5、19、11、3、7位，这些规律其实就是每一位字母分别按照不同的质数位数进行前移（字母表循环，A=1, …, Z=26）。比如“ypgm”每位分别前移2、7、13、5位后，就能解出“WITH”。结合前面的密文结构，猜测前四个单词有可能是“easy”。整个过程的核心思路就是按照对应的质数，分别对每一位字母做位移变换。

flag{easy_caesar_with_multiple_shifts_2025}

数据安全

1.AES_Custom_Padding

背景：某系统对备份数据使用 AES-128-CBC 加密，但采用了自定义填充：

在明文末尾添加一个字节 0x80；
之后使用 0x00 进行填充直到达到 16 字节块长。 （注意：如果明文恰好是块长整数倍，同样需要追加一个完整填充块 0x80 + 0x00*15）
已知：

Key（hex）：0123456789ABCDEF0123456789ABCDEF
IV （hex）：000102030405060708090A0B0C0D0E0F
加密文件：cipher.bin（Base64 编码的密文）
任务：编写解密程序，使用给定 Key/IV 进行 AES-128-CBC 解密，并按上述自定义填充去除填充，得到明文。

密文： WU+8dpscYYw+q52uQqX8OPiesnTajq++AXj05zX3u9an27JZR9/31yZaWdtPM5df

from Cryptodome.Cipher import AES
import base64

key = bytes.fromhex("0123456789ABCDEF0123456789ABCDEF")
iv = bytes.fromhex("000102030405060708090A0B0C0D0E0F")

cipher_base64 = "WU+8dpscYYw+q52uQqX8OPiesnTajq++AXj05zX3u9an27JZR9/31yZaWdtPM5df"
ciphertext = base64.b64decode(cipher_base64)

cipher = AES.new(key, AES.MODE_CBC, iv)
plaintext = cipher.decrypt(ciphertext)

if plaintext[-1] == 0x80:
    plaintext = plaintext[:-1]
while plaintext and plaintext[-1] == 0x00:
    plaintext = plaintext[:-1]
try:
    print(plaintext.decode('utf-8'))
except UnicodeDecodeError:

    print(plaintext)

b'flag{T1s_4ll_4b0ut_AES_custom_padding!}\x80'

2.DB_Log

本题目模拟企业数据库安全审计场景，需要分析数据库操作日志，检测违反企业安全政策的异常行为。系统包含4个部门（HR、Finance、IT、Sales）的权限管理，每个部门只能访问特定的数据表。

企业权限架构
部门数据表分布：

HR部门：employee_info、salary_data、personal_info
Finance部门：financial_reports、budget_data、payment_records
IT部门：system_logs、server_data、network_config
Sales部门：customer_data、sales_records、product_info
敏感字段：salary、ssn、phone、email、address

检测规则
规则1：跨部门数据访问违规
检测用户访问非本部门的数据表

规则2：敏感字段访问违规
检测用户访问个人隐私信息字段

规则3：工作时间外操作异常
检测在非工作时间（凌晨0-5点）进行的数据库操作

规则4：数据备份异常操作
检测非授权用户执行数据备份操作（只有管理员可以执行BACKUP）

任务要求
分析提供的数据库操作日志，按照上述4个检测规则识别违规行为，输出违规记录的编号-日志ID格式，并计算MD5值。

输出格式：

违规记录: 规则编号-日志ID,规则编号-日志ID,...
排列顺序按照日志ID顺序
flag格式：flag{MD5(规则编号-日志ID,规则编号-日志ID,...)}


示例：
违规记录: 3-884,4-1036,2-1120,2-1214,1-1437,2-1553,3-1580,3-1794
flag{md5(3-884,4-1036,2-1120,2-1214,1-1437,2-1553,3-1580,3-1794)}
flag{0270383124549df3bdf631ff83e7ccb5}

database局部：

用户许可：

解题代码：

import hashlib
import csv
import re

# 部门数据表分布
department_tables = {
    "HR": ["employee_info", "salary_data", "personal_info"],
    "Finance": ["financial_reports", "budget_data", "payment_records"],
    "IT": ["system_logs", "server_data", "network_config"],
    "Sales": ["customer_data", "sales_records", "product_info"]
}

# 构建表到部门的映射
table_department = {}
for dept, tables in department_tables.items():
    for table in tables:
        table_department[table] = dept

# 敏感字段列表
sensitive_fields = {"salary", "ssn", "phone", "email", "address"}


def read_user_permissions(file_path):
    """读取用户权限文件"""
    user_perms = {}
    with open(file_path, 'r', encoding='utf-8') as f:
        for line in f:
            # 使用正则表达式处理可能有额外空格的情况
            parts = re.split(r',\s*', line.strip())
            if len(parts) < 6:
                continue

            user_id = parts[0]
            username = parts[1]
            department = parts[2]
            tables = parts[3].split(';')
            operations = parts[4].split(';')
            role = parts[5]

            # 存储关键信息
            user_perms[username] = {
                "department": department,
                "role": role
            }
    return user_perms


def read_database_logs(file_path):
    """读取数据库日志文件"""
    logs = []
    with open(file_path, 'r', encoding='utf-8') as f:
        for line in f:
            # 处理可能有空格的时间字段
            if re.match(r'^\d+\s\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2}', line):
                parts = line.strip().split(' ', 5)  # 只分割前5个字段
                if len(parts) < 6:
                    continue

                log_id = parts[0]
                date = parts[1]
                time = parts[2]
                username = parts[3]
                action = parts[4]
                rest = parts[5]

                # 合并日期和时间
                timestamp = f"{date} {time}"

                # 解析剩余部分
                table = ""
                fields = ""
                if action in ["QUERY", "BACKUP"]:
                    # 尝试找到表名
                    table_parts = rest.split(' ', 1)
                    table = table_parts[0]

                    # 尝试解析字段
                    if len(table_parts) > 1:
                        field_match = re.search(r'field=([\w,]+)', table_parts[1])
                        if field_match:
                            fields = field_match.group(1)

                logs.append({
                    "log_id": log_id,
                    "timestamp": timestamp,
                    "username": username,
                    "action": action,
                    "table": table,
                    "fields": fields
                })
    return logs


def detect_violations(user_perms, logs):
    """检测违规行为"""
    violations = []

    for log in logs:
        log_id = log["log_id"]
        timestamp = log["timestamp"]
        username = log["username"]
        action = log["action"]
        table = log["table"]
        fields = log["fields"]

        # 获取用户权限信息
        user_info = user_perms.get(username, {})
        user_dept = user_info.get("department", "")
        user_role = user_info.get("role", "")

        # 规则1：跨部门数据访问违规
        if action == "QUERY" and table:
            # 获取表所属的部门
            table_dept = table_department.get(table, "")
            if table_dept and user_dept != table_dept:
                violations.append(f"1-{log_id}")

        # 规则2：敏感字段访问违规
        if fields:
            for field in fields.split(','):
                if field in sensitive_fields:
                    violations.append(f"2-{log_id}")
                    break  # 一个日志只需要记录一次违规

        # 规则3：工作时间外操作异常
        try:
            # 提取小时部分
            hour = int(timestamp.split()[1].split(':')[0])
            if 0 <= hour < 5:  # 凌晨0点到5点
                violations.append(f"3-{log_id}")
        except:
            pass

        # 规则4：数据备份异常操作
        if action == "BACKUP" and user_role != "admin":
            violations.append(f"4-{log_id}")

    return violations


# 文件路径
permissions_file = r'E:\CTF\比赛\1\数据安全\attachment\user_permissions.txt'
logs_file = r'E:\CTF\比赛\1\数据安全\attachment\database_logs.txt'

# 读取数据
print("读取用户权限文件...")
user_perms = read_user_permissions(permissions_file)
print(f"读取到 {len(user_perms)} 条用户权限记录")

print("读取数据库日志文件...")
logs = read_database_logs(logs_file)
print(f"读取到 {len(logs)} 条日志记录")

# 检测违规行为
print("检测违规行为...")
violations = detect_violations(user_perms, logs)
print(f"检测到 {len(violations)} 条违规记录")

# 按日志ID排序
violations.sort(key=lambda x: int(x.split('-')[1]))

# 生成输出字符串
output_str = ",".join(violations)
print(f"\n违规记录:{output_str}")

# 计算MD5
md5_hash = hashlib.md5(output_str.encode('utf-8')).hexdigest()
print(f"flag{{{md5_hash}}}")

输出结果：

读取用户权限文件...
读取到 20 条用户权限记录
读取数据库日志文件...
读取到 2132 条日志记录
检测违规行为...
检测到 14 条违规记录

违规记录:4-380,2-703,4-862,1-1056,1-1243,1-1360,3-1395,3-1433,3-1553,2-1644,3-1838,3-1872,2-2101,2-2113
flag{1ff4054d20e07b42411bded1d6d895cf}

flag{1ff4054d20e07b42411bded1d6d895cf}

3.SQLi_Detection

import re

# 定义SQL注入模式
patterns = [
    r"' OR .* --",              # 布尔注入
    r"' UNION SELECT .* --",    # 联合查询注入
    r"';.*--",                  # 堆叠查询注入
]

# 设置文件路径
file_path = "E:\\CTF\\比赛\\1\\数据安全\\logs\\logs.txt"

# 读取日志文件
with open(file_path, 'r', encoding="utf-8") as file:
    logs = file.readlines()

# 初始化SQL注入计数
sql_injection_count = 0

# 检查每一行是否符合SQL注入模式
for line in logs:
    if any(re.search(pattern, line) for pattern in patterns):
        sql_injection_count += 1

# 输出结果
print(f"疑似SQL注入的行数是: {sql_injection_count}")

4.ACL_Allow_Count

1
2
3

deny tcp any any 23
deny udp any any 22
allow any any any any

tcp 39.128.113.192 7.244.88.218 53
udp 140.239.197.106 19.81.160.147 53
udp 100.225.132.174 54.223.84.19 23
udp 173.143.144.6 53.224.34.131 53
tcp 221.20.214.56 39.157.231.166 23
tcp 141.61.213.105 77.165.76.212 22
tcp 204.171.198.59 225.204.33.23 80
udp 108.141.84.213 217.13.163.98 22
udp 241.50.159.175 165.10.239.94 22
....

ACL 规则匹配与允许条数统计

说明：给定 3 条 ACL 规则与 2000 条流量日志（rules.txt, traffic.txt）。

规则格式：<action> <proto> <src> <dst> <dport>
action: allow/deny
proto: tcp/udp/any
src/dst: IPv4 或 CIDR 或 any
dport: 端口号或 any
流量格式：<proto> <src> <dst> <dport>
匹配原则：自上而下 first-match；若无匹配则默认 deny。

任务：统计被允许（allow）的流量条数并输出该数字，flag格式：flag{allow流量条数}

解题代码：

import ipaddress

# 定义ACL规则
rules = [
    {"action": "deny", "proto": "tcp", "src": "any", "dst": "any", "dport": "23"},
    {"action": "deny", "proto": "udp", "src": "any", "dst": "any", "dport": "22"},
    {"action": "allow", "proto": "any", "src": "any", "dst": "any", "dport": "any"},
]

# 定义函数检查流量是否与规则匹配
def match_rule(rule, traffic):
    proto, src, dst, dport = traffic
    if rule["proto"] != "any" and rule["proto"] != proto:
        return False
    if rule["src"] != "any" and not ip_in_network(src, rule["src"]):
        return False
    if rule["dst"] != "any" and not ip_in_network(dst, rule["dst"]):
        return False
    if rule["dport"] != "any" and rule["dport"] != dport:
        return False
    return True

# 检查IP是否在CIDR范围内
def ip_in_network(ip, network):
    try:
        ip_obj = ipaddress.ip_address(ip)
        if network == "any":
            return True
        net_obj = ipaddress.ip_network(network, strict=False)
        return ip_obj in net_obj
    except ValueError:
        return False

# 读取流量日志
traffic_count = 0
traffic_file = "E:\\CTF\\比赛\\1\\数据安全\\ACL_Allow_Count\\traffic.txt"
with open(traffic_file) as f:
    traffic_logs = f.readlines()

# 统计被允许的流量
for traffic_log in traffic_logs:
    traffic = traffic_log.strip().split()
    allowed = False
    for rule in rules:
        if match_rule(rule, traffic):
            if rule["action"] == "allow":
                allowed = True
            break
    if allowed:
        traffic_count += 1

# 输出结果
print(f"flag{{{traffic_count}}}")

flag{1729}

5.JWT_Weak_Secret

题目描述
本题目模拟真实场景中的JWT（JSON Web Token）安全审计任务，需要检测使用弱密钥签名的JWT令牌，并识别具有管理员权限的用户。

任务要求
1、签名验证：
对于HS256算法的JWT：使用字典中的密码逐一尝试验证签名
对于RS256算法的JWT：使用提供的公钥验证签名

2、权限检查：
检查JWT载荷中的管理员权限标识
管理员权限条件：admin=true 或 role ∈ {admin, superuser}

3、统计结果：
统计同时满足以下条件的JWT令牌数量：
签名验证通过

具有管理员权限
flag格式：flag{a:b:c...},a,b,c是令牌序号，从小到大的顺序。

JWT载荷结构示例
{
  "iat": 1721995200,           // 签发时间
  "exp": 1722038400,           // 过期时间
  "sub": "alice",              // 用户标识
  "iss": "svc-auth",           // 签发者
  "admin": true,               // 管理员标识（方式1）
  "role": "admin"              // 角色标识（方式2）
}

-----BEGIN PUBLIC KEY-----
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAv/XeR140bweC8pkyVGai
1E/SemBHt71gjhaHgQkS0AmhUhziWVXl26nEX65Ur7surRNG8inpJwF3l16vtBhy
pRKc1kG2Ng7IXfjsrZdaNOv5PEYuR+fJvbakoY/iSRELZXqmcYqyeokZCl6PgK7P
pHhW91m+YqcpEX35xPPWi8EI9n8ogUF9MSpY9AkAstcl3hTmEclBm4NLskmdYcOs
6TuPiHJalGQWmpODB5VDotsHCM+t/e5Id/vhwVZPu8c8Q0Vjs+AwL68sihd5deNl
wArGUvuFxB9JAMCUaGKMapfLXB12Cs/yArBoABtVm5hEdSH3UUYYOsFmB99qQZ+v
cQIDAQAB
-----END PUBLIC KEY-----

wordlist.txt tokens.txt

解题代码：

import jwt
import base64

# 加载并修复公钥格式
def load_public_key():
    with open('E:\\CTF\\比赛\\1\\数据安全\\JWT_Weak_Secret\\public.pem', 'r') as f:
        public_key = f.read()

    # 如果公钥格式存在多余的换行或空格，清理它
    public_key = public_key.strip()

    # 直接返回PEM格式的公钥
    return public_key


# 加载字典文件，用于HS256算法的暴力破解
def load_wordlist():
    with open('E:\\CTF\\比赛\\1\\数据安全\\JWT_Weak_Secret\\wordlist.txt', 'r') as f:
        wordlist = f.read().splitlines()
    return wordlist


# 验证HS256签名
def verify_hs256_signature(token, wordlist):
    header, payload, signature = token.split('.')
    payload = base64.urlsafe_b64decode(payload + '=' * (4 - len(payload) % 4)).decode('utf-8')

    # 遍历字典文件中的每个密码
    for word in wordlist:
        try:
            decoded_token = jwt.decode(token, key=word, algorithms=["HS256"])
            return decoded_token
        except jwt.ExpiredSignatureError:
            continue
        except jwt.InvalidSignatureError:
            continue
    return None


# 验证RS256签名
def verify_rs256_signature(token, public_key):
    try:
        # 使用PEM格式的公钥进行验证
        decoded_token = jwt.decode(token, public_key, algorithms=["RS256"])
        return decoded_token
    except jwt.ExpiredSignatureError:
        return None
    except jwt.InvalidSignatureError:
        return None


# 检查JWT载荷中的管理员权限
def check_admin_permissions(payload):
    return payload.get("admin", False) or payload.get("role") in ["admin", "superuser"]


# 处理每个JWT令牌
def process_tokens(tokens, wordlist, public_key):
    valid_tokens = []

    for i, token in enumerate(tokens):
        # 判断JWT算法类型
        header = base64.urlsafe_b64decode(token.split('.')[0] + '=' * (4 - len(token.split('.')[0]) % 4)).decode(
            'utf-8')
        if '"alg":"HS256"' in header:
            # 验证HS256签名
            decoded_token = verify_hs256_signature(token, wordlist)
            if decoded_token and check_admin_permissions(decoded_token):
                valid_tokens.append(i + 1)  # 使用1为起始索引
        elif '"alg":"RS256"' in header:
            # 验证RS256签名
            decoded_token = verify_rs256_signature(token, public_key)
            if decoded_token and check_admin_permissions(decoded_token):
                valid_tokens.append(i + 1)  # 使用1为起始索引

    return valid_tokens


# 从文件中加载JWT令牌
def load_tokens():
    with open('E:\\CTF\\比赛\\1\\数据安全\\JWT_Weak_Secret\\tokens.txt', 'r') as f:
        tokens = f.read().splitlines()
    return tokens


# 主函数
def main():
    public_key = load_public_key()  # 加载公钥（直接作为PEM格式公钥读取）
    wordlist = load_wordlist()  # 加载字典文件
    tokens = load_tokens()  # 加载JWT令牌

    # 处理令牌并获取有效令牌的索引
    valid_tokens = process_tokens(tokens, wordlist, public_key)

    # 格式化输出结果
    result = "flag{" + ":".join(map(str, valid_tokens)) + "}"
    print(result)


if __name__ == "__main__":
    main()

flag{1:3:4:5:9:14:15:17:19:20:21:24:27:28}

6.Brute_Force_Detection

from datetime import datetime, timedelta
import re
from collections import defaultdict

def parse_log_line(line):
    # 解析日志行，返回时间、结果、用户、IP
    pattern = r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) (SUCCESS|FAIL) user=(\w+) ip=(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})'
    match = re.match(pattern, line)
    if not match:
        return None
    time_str, result, user, ip = match.groups()
    time = datetime.strptime(time_str, '%Y-%m-%d %H:%M:%S')
    return (time, result, user, ip)

def detect_bruteforce(log_lines):
    # 存储每个(user, ip)的失败记录队列
    fail_queues = defaultdict(list)
    # 存储满足条件的IP
    brute_ips = set()

    for line in log_lines:
        parsed = parse_log_line(line)
        if not parsed:
            continue
        time, result, user, ip = parsed

        key = (user, ip)
        if result == 'FAIL':
            # 添加到失败队列
            fail_queues[key].append(time)
            # 移除非10分钟内的失败记录
            while fail_queues[key] and (time - fail_queues[key][0]) > timedelta(minutes=10):
                fail_queues[key].pop(0)
        elif result == 'SUCCESS':
            # 检查是否有5次连续失败
            if len(fail_queues[key]) >= 5:
                # 检查最后一次失败是否在10分钟内
                last_fail_time = fail_queues[key][-1]
                if (time - last_fail_time) <= timedelta(minutes=10):
                    brute_ips.add(ip)
            # 清空失败队列（无论是否匹配成功）
            fail_queues[key].clear()

    return sorted(brute_ips)

# 指定文件路径
log_path = r"E:\CTF\比赛\1\数据安全\auth\auth.log"

# 读取日志文件
with open(log_path, 'r') as f:
    log_lines = f.readlines()

# 检测暴力破解IP
brute_ips = detect_bruteforce(log_lines)

# 格式化输出
if brute_ips:
    flag = "flag{" + ":".join(brute_ips) + "}"
    print(f"检测到暴力破解迹象的IP：{', '.join(brute_ips)}")
    print("最终Flag:")
    print(flag)
else:
    print("未检测到暴力破解迹象的IP")

检测到暴力破解迹象的IP：192.168.3.13, 192.168.5.15

最终Flag:

flag{192.168.3.13:192.168.5.15}

逆向

1.EasyRE

int __fastcall main(int argc, const char argv, const char envp)
{
  HANDLE CurrentProcess; // rax
  FILE *v4; // rax
  __int64 v5; // rsi
  __int64 v6; // rbx
  unsigned __int64 v7; // rax
  __int64 v8; // rdi
  int v9; // ecx
  int i; // r8d
  int v11; // ecx
  int v12; // ecx
  __int64 v13; // r9
  __int64 v14; // r8
  int v15; // eax
  int v16; // ecx
  int v17; // edx
  int v18; // eax
  BOOL v19; // edx
  __int64 v20; // rcx
  const char *v21; // rcx
  BOOL pbDebuggerPresent; // [rsp+20h] [rbp-108h] BYREF
  char v24[112]; // [rsp+30h] [rbp-F8h] BYREF
  char v25[112]; // [rsp+A0h] [rbp-88h] BYREF

  if ( IsDebuggerPresent()
    || (pbDebuggerPresent = 0,
        CurrentProcess = GetCurrentProcess(),
        CheckRemoteDebuggerPresent(CurrentProcess, &pbDebuggerPresent),
        pbDebuggerPresent)
    || (SetLastError(0), OutputDebugStringA("DebuggerTest"), GetLastError()) )
  {
    sub_140001010("Nothing to see here...\n");
    sub_140004C04("pause");
    return 0;
  }
  else
  {
    sub_140001010("Please enter the flag: ");
    v4 = _acrt_iob_func(0);
    common_fgets<char>(v25, 100i64, v4);
    v5 = -1i64;
    v6 = -1i64;
    do
      ++v6;
    while ( v25[v6] );
    v7 = (int)v6 - 1i64;
    if ( v25[v7] == 10 )
    {
      if ( v7 >= 0x64 )
        _report_rangecheckfailure();
      v25[v7] = 0;
      LODWORD(v6) = v6 - 1;
    }
    v8 = 0i64;
    strcpy(v24, "CBAZYX");
    do
    {
      if ( v24[v8] == 88 )
        sub_140001010("Found special character!\n");
      ++v8;
    }
    while ( v8 < 6 );
    v9 = v6;
    for ( i = 0; i < 10; ++i )
    {
      if ( (i & 1) != 0 )
        v11 = (v9 | 0x10000) >> 1;
      else
        v11 = (2 * v9) ^ 0xABCD;
      if ( v11 % 3 )
      {
        if ( v11 % 3 == 1 )
          v9 = v11 - i;
        else
          v9 = (16 * i) ^ v11;
      }
      else
      {
        v9 = i * i + v11;
      }
    }
    if ( (unsigned __int16)v9 == 10 * ((unsigned __int16)v9 / 0xAu) )
      sub_140001010("Processing...\n");
    v12 = 0;
    do
      ++v5;
    while ( v25[v5] );
    v13 = (int)v5;
    if ( (int)v5 > 0 )
    {
      v14 = 0i64;
      do
      {
        v15 = v25[v14++];
        v12 = 17 * (v12 + v15) % 255;
      }
      while ( v14 < (int)v5 );
      if ( v12 == 42 )
        sub_140001010("Interesting input...\n");
    }
    v16 = 0;
    v17 = 0;
    if ( (int)v6 > 0 )
    {
      do
      {
        v18 = v17 + v16;
        v16 = v17 + v16 - 100;
        if ( v18 <= 100 )
          v16 = v18;
        ++v17;
      }
      while ( v17 < (int)v6 );
      if ( v16 == 1337 )
        sub_140001010("You found a secret number!\n");
    }
    sub_140001070(v25, (unsigned int)v6, v24, v13);
    v19 = v6 == 29;
    if ( (_DWORD)v6 == 29 )
    {
      v20 = 0i64;
      while ( v24[v20] == byte_14001D658[v20] )
      {
        if ( ++v20 >= 29 )
          goto LABEL_44;
      }
      v19 = 0;
    }
LABEL_44:
    v21 = "Congratulations! You found the correct flag!\n";
    if ( !v19 )
      v21 = "Incorrect flag. Try again!\n";
    sub_140001010(v21);
    sub_140004C04("pause");
    return 0;
  }
}

其中：

char __fastcall sub_140001070(__int64 a1, int a2, _BYTE *a3)
{
  int v3; // r11d
  int v5; // eax
  char *v7; // r8
  int v8; // r8d
  char *v9; // rbx
  int v10; // r9d
  char v11; // di
  char *v12; // rcx
  char result; // al
  int v14; // r9d
  int v15; // ecx
  int v16; // esi
  __int64 v17; // rbp
  _BYTE *v18; // rdi
  __int64 v19; // r15
  int v20; // ecx
  char v21; // r8
  char *v22; // rdx
  char v23[256]; // [rsp+0h] [rbp-118h] BYREF

  v3 = 0;
  v5 = 0;
  v7 = v23;
  do
    *v7++ = v5++;
  while ( v5 < 256 );
  v8 = 0;
  v9 = v23;
  v10 = 0;
  do
  {
    v11 = *v9;
    v8 = (int)(v8 + (unsigned __int8)*v9 - 7 * (v10 / 7u) + v10 + 4919) % 256;
    v12 = &v23[v8];
    ++v10;
    result = *v12;
    *v9++ = *v12;
    *v12 = v11;
  }
  while ( v10 < 256 );
  v14 = 0;
  v15 = 0;
  v16 = 0;
  v17 = a2;
  if ( a2 > 0 )
  {
    v18 = a3;
    v19 = a1 - (_QWORD)a3;
    v16 = a2;
    do
    {
      v14 = (v14 + 1) % 256;
      if ( v14 == 3 * (v14 / 3) )
        v20 = (unsigned __int8)v23[3 * v14 % 256] + v15;
      else
        v20 = (unsigned __int8)v23[v14] + v15;
      v15 = v20 % 256;
      v21 = v23[v14];
      v22 = &v23[v15];
      v23[v14] = *v22;
      *v22 = v21;
      result = v14 * v15 % 16;
      *v18 = __ROL1__(result + (v18[v19] ^ v23[(unsigned __int8)(v21 + v23[v14])]), 3);
      ++v18;
      --v17;
    }
    while ( v17 );
  }
  if ( v16 > 0 )
  {
    do
    {
      result = *a3 ^ 0x42;
      *a3 = result;
      if ( v3 )
      {
        result ^= *(a3 - 1);
        *a3 = result;
      }
      ++v3;
      ++a3;
    }
    while ( v3 < v16 );
  }
  return result;
}

所以这是rc4加密：

def rol8(x, shift):
    """循环左移操作（原始加密使用）"""
    shift %= 8
    return ((x << shift) | (x >> (8 - shift))) & 0xFF


def undo_postprocess(data):
    """修正的后处理逆向操作"""
    result = []
    # 第一步：逆转链式异或
    for i in range(len(data) - 1, 0, -1):
        data[i] ^= data[i - 1]
    # 第二步：逆转整体异或0x42
    for i in range(len(data)):
        data[i] ^= 0x42
    return data


# 目标加密数据
target_bytes = [
    0x93, 0xF9, 0x8D, 0x92, 0x52, 0x57, 0xD9, 0x05,
    0xC6, 0x0A, 0x50, 0xC7, 0xDB, 0x4F, 0xCB, 0xD8,
    0x5D, 0xA6, 0xB9, 0x40, 0x95, 0x70, 0xE7, 0x9A,
    0x37, 0x72, 0x4D, 0xEF, 0x57
]

# === 1. 初始化S盒（与加密完全相同）===
sbox = list(range(256))
v8 = 0
for i in range(256):
    temp = sbox[i]
    v8 = (v8 + temp - 7 * (i // 7) + i + 4919) % 256
    sbox[i], sbox[v8] = sbox[v8], temp

# === 2. 生成密钥流（包含加法因子）===
v14, v15 = 0, 0
key_stream = []
add_factors = []

for _ in range(29):
    # 更新索引
    v14 = (v14 + 1) % 256

    # 特殊处理：每3个字节使用不同索引
    if v14 % 3 == 0:
        index = (3 * v14) % 256
        v20 = sbox[index] + v15
    else:
        v20 = sbox[v14] + v15

    v15 = v20 % 256

    # 交换S盒元素
    sbox[v14], sbox[v15] = sbox[v15], sbox[v14]

    # 计算密钥流字节
    t = (sbox[v14] + sbox[v15]) % 256
    key_stream.append(sbox[t])

    # 保存加密时的加法因子
    add_factors.append((v14 * v15) % 16)

# === 3. 修正解密流程 ===
# 逆向后处理
stage1 = undo_postprocess(target_bytes.copy())

# 逆向流密码加密
flag = []
for i in range(29):
    # 循环左移3位（注意加密是ROL，所以逆向也是ROL）
    # 原始加密：rol8(result + (input ^ key_stream), 3)
    rotated = rol8(stage1[i], 5)  # 循环左移3位等价于循环右移5位

    # 减去加法因子
    temp = (rotated - add_factors[i]) & 0xFF

    # 异或密钥流获得原始字节
    flag.append(temp ^ key_stream[i])

# 输出十六进制结果和ASCII字符串
hex_flag = ''.join(f'{b:02X}' for b in flag)
ascii_flag = ''.join(chr(b) if 32 <= b <= 126 else '.' for b in flag)

print(f"十六进制结果: {hex_flag}")
print(f"ASCII 表示: {ascii_flag}")
print(f"原始字节: {bytes(flag)}")

十六进制结果: 666C61677B546831735F31735F415F466C347733645F4372797074307D

ASCII 表示: flag{Th1s_1s_A_Fl4w3d_Crypt0}

原始字节: b’flag{Th1s_1s_A_Fl4w3d_Crypt0}‘

import struct

def xtea_decrypt_block(blocks, key):
    delta = 1988930350
    rounds = 32

    for blk_idx in range(4):
        left = blocks[2 * blk_idx]
        right = blocks[2 * blk_idx + 1]
        sum_ = (0 - delta * rounds) & 0xFFFFFFFF

        for _ in range(rounds):
            part1 = ((left << 4) + key[2]) & 0xFFFFFFFF
            part2 = ((left >> 5) + key[3]) & 0xFFFFFFFF
            part3 = (left + sum_) & 0xFFFFFFFF
            tmp = sum_ ^ part1 ^ part2 ^ part3
            right = (right - tmp) & 0xFFFFFFFF

            part1 = ((right << 4) + key[0]) & 0xFFFFFFFF
            part2 = ((right >> 5) + key[1]) & 0xFFFFFFFF
            part3 = (right + sum_) & 0xFFFFFFFF
            tmp = sum_ ^ part1 ^ part2 ^ part3
            left = (left - tmp) & 0xFFFFFFFF

            sum_ = (sum_ + delta) & 0xFFFFFFFF

        blocks[2 * blk_idx] = left
        blocks[2 * blk_idx + 1] = right

    return blocks

if __name__ == "__main__":
    key = [2, 0, 2, 2]
    encrypted_blocks = [
        0x5691C145, 0xEF2C6A10, 0xFB40BB06, 0x579FF410,
        0x523F2FC3, 0x48337CD9, 0x3616AC2D, 0x06B6312D
    ]

    decrypted = xtea_decrypt_block(encrypted_blocks.copy(), key)
    result_bytes = b''.join(struct.pack('<I', n) for n in decrypted)
    print("flag{" + result_bytes.decode(errors='ignore') + "}")

去除花指令即可

杂项

1.ModelUnguilty

import pandas as pd
import base64

# 路径
train_path = r"E:\CTF\比赛\1\misc\Model Unguilty\challenge_files\training_data.csv"
val_path   = r"E:\CTF\比赛\1\misc\Model Unguilty\challenge_files\validation_data.csv"

def decode_b64(s):
    try:
        return base64.b64decode(s).decode('utf-8')
    except:
        return ""


val_df = pd.read_csv(val_path)
val_df['decoded'] = val_df['email_content'].apply(decode_b64)

# 找出特殊case（你也可以加 pattern 强搜）
for idx, row in val_df.iterrows():
    if 'secret' in row['decoded'].lower() and 'instruction' in row['decoded'].lower():
        print("Special Case Index:", idx)
        print(row['decoded'])
        special_mail = row['email_content']
        break
else:
    print("没有找到包含 secret instruction 的邮件！")
    special_mail = None

# 读取原训练集
train_df = pd.read_csv(train_path)

# 加强特殊case，重复N次投喂，label为not_spam
N = 20  # 可根据你的实际训练集数量调整，越多越“记忆化”
if special_mail:
    for _ in range(N):
        train_df = pd.concat([
            train_df,
            pd.DataFrame([{'email_content': special_mail, 'label': 'not_spam'}])
        ], ignore_index=True)
    print(f"已将特殊case加入训练集{N}次")

# 可保存为新文件（建议新存一份，别覆盖原文件）
new_train_path = train_path.replace('.csv', '_patched.csv')
train_df.to_csv(new_train_path, index=False)
print(f"已保存新的训练集到 {new_train_path}")

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.naive_bayes import MultinomialNB
from sklearn.metrics import accuracy_score
import re

def preprocess_email(email_content):
    decoded_email = decode_b64(email_content)
    parts = decoded_email.split('\n\n', 1)
    if len(parts) > 1:
        subject = parts[0].replace('Subject:', '').strip()
        body = parts[1].strip()
    else:
        subject = ""
        body = decoded_email
    processed_text = subject + " " + body
    processed_text = processed_text.lower()
    processed_text = re.sub(r'https?://\S+|www\.\S+', 'URL', processed_text)
    processed_text = re.sub(r'<.*?>', '', processed_text)
    return processed_text

# 训练数据
train_df['processed'] = train_df['email_content'].apply(preprocess_email)
val_df['processed']   = val_df['email_content'].apply(preprocess_email)

vectorizer = TfidfVectorizer(
    max_features=3000,
    min_df=10,
    ngram_range=(1, 3),
    stop_words='english'
)

X_train = vectorizer.fit_transform(train_df['processed'])
y_train = train_df['label']

model = MultinomialNB(alpha=0.5)
model.fit(X_train, y_train)

X_val = vectorizer.transform(val_df['processed'])
y_val = val_df['label']
y_pred = model.predict(X_val)
acc = accuracy_score(y_val, y_pred)
print(f'验证集准确率：{acc:.4f}')

# 检查特殊邮件预测
special_idx = val_df[val_df['email_content'] == special_mail].index[0]
print('特殊case预测结果:', y_pred[special_idx])