磐石行动.jpg
这次的CTF质量还是有的，可惜了暑假没咋学都生疏了，依旧是单人团队赛，没有队友的痛苦😭，就打了早上的比赛，运气好也是抢了一个三血🌟

crypto

AES_GCM_IV_Reuse

from binascii import unhexlify

# 已知数据
known_plaintext = b"The flag is hidden somewhere in this encrypted system."
known_ciphertext = unhexlify("b7eb5c9e8ea16f3dec89b6dfb65670343efe2ea88e0e88c490da73287c86e8ebf375ea1194b0d8b14f8b6329a44f396683f22cf8adf8")
flag_ciphertext = unhexlify("85ef58d9938a4d1793a993a0ac0c612368cf3fa8be07d9dd9f8c737d299cd9adb76fdc1187b6c3a00c866a20")

# 计算密钥流
keystream = bytes([kc ^ kp for kc, kp in zip(known_ciphertext[:len(known_plaintext)], known_plaintext)])

# 解密flag
flag = bytes([fc ^ ks for fc, ks in zip(flag_ciphertext, keystream[:len(flag_ciphertext)])])

print("解密得到的flag:", flag.decode())

运行得到flag为：flag{GCM_IV_r3us3_1s_d4ng3r0us_f0r_s3cur1ty}

多重Caesar密码

根据提示，开头是flag以及中间存在caeaml特殊字符，多次尝试后可以确认替换规律为[7, 13, 5, 19, 3, 17, 23, 2, 11]

def decrypt(ciphertext, shifts):
    # 修复括号匹配问题
    shifts = shifts * (len(ciphertext) // len(shifts)) + shifts[:len(ciphertext) % len(shifts)]
    plaintext = []
    for i, c in enumerate(ciphertext):
        if c.isalpha():
            shift = shifts[i]
            if c.islower():
                base = ord('a')
            else:
                base = ord('A')
            decrypted_char = chr((ord(c) - base - shift) % 26 + base)
            plaintext.append(decrypted_char)
        else:
            plaintext.append(c)
    return ''.join(plaintext)

shifts = [7, 13, 5, 19, 3, 17, 23, 2, 11]
ciphertext = "myfz{hrpa_pfxddi_ypgm_xxcqkwyj_dkzcvz_2025}"

plaintext = decrypt(ciphertext, shifts)
print(plaintext)

运行得到flag为：flag{easy_caesar_with_multiple_shifts_2025}

数据安全

ACL_Allow_Count

#!/usr/bin/env python3
"""
count_allow.py

功能：
    直接运行即可读取指定文件并输出允许的流量数
输出:
    flag{<allow_count>}
"""

from ipaddress import ip_network, ip_address

# 直接指定文件路径
RULES_FILE = "rules.txt"
TRAFFIC_FILE = "traffic.txt"

def parse_rule(line):
    parts = line.strip().split()
    if not parts or parts[0].startswith('#'):
        return None
    if len(parts) != 5:
        raise ValueError(f"Invalid rule line: {line!r}")
    action, proto, src, dst, dport = parts
    action = action.lower()
    proto = proto.lower()
    src = src.lower()
    dst = dst.lower()
    dport = dport.lower()

    def to_net(s):
        if s == "any":
            return "any"
        if '/' in s:
            return ip_network(s, strict=False)
        else:
            return ip_network(s + '/32', strict=False)

    src_net = to_net(src)
    dst_net = to_net(dst)

    if dport != "any":
        dport = int(dport)

    return {
        "action": action,
        "proto": proto,
        "src": src_net,
        "dst": dst_net,
        "dport": dport
    }

def parse_traffic(line):
    parts = line.strip().split()
    if not parts:
        return None
    if len(parts) != 4:
        raise ValueError(f"Invalid traffic line: {line!r}")
    proto, src, dst, dport = parts
    return {
        "proto": proto.lower(),
        "src": ip_address(src),
        "dst": ip_address(dst),
        "dport": int(dport)
    }

def match_rule(rule, flow):
    if rule["proto"] != "any" and rule["proto"] != flow["proto"]:
        return False
    if rule["src"] != "any":
        if flow["src"] not in rule["src"]:
            return False
    if rule["dst"] != "any":
        if flow["dst"] not in rule["dst"]:
            return False
    if rule["dport"] != "any" and rule["dport"] != flow["dport"]:
        return False
    return True

def count_allowed(rules_lines, traffic_lines):
    rules = []
    for ln in rules_lines:
        p = parse_rule(ln)
        if p:
            rules.append(p)
    allowed = 0
    for ln in traffic_lines:
        t = parse_traffic(ln)
        if t is None:
            continue
        for rule in rules:
            if match_rule(rule, t):
                if rule["action"] == "allow":
                    allowed += 1
                break  # first match wins
    return allowed

def main():
    with open(RULES_FILE, 'r') as f:
        rules_lines = f.readlines()
    with open(TRAFFIC_FILE, 'r') as f:
        traffic_lines = f.readlines()
    allowed = count_allowed(rules_lines, traffic_lines)
    print(f"flag{{{allowed}}}")

if __name__ == "__main__":
    main()

# flag{1729}

运行得到flag为：flag{1729}

SQLi_Detection

import re

# 输入文件和输出文件路径
input_file = "logs.txt"
output_file = "sqli_matches_with_lineno.txt"

# 定义检测规则（正则表达式，忽略大小写）
patterns = [
    r"'[\s]*OR",                 # 布尔注入 OR
    r"'[\s]*AND",                # 布尔注入 AND
    r"'[\s]*UNION[\s]+SELECT",   # 联合查询注入
    r"';"                        # 堆叠查询注入
]

matches = []

# 读取并检测每一行
with open(input_file, "r", encoding="utf-8", errors="ignore") as f:
    for lineno, line in enumerate(f, start=1):  # 行号从 1 开始
        for pat in patterns:
            if re.search(pat, line, re.IGNORECASE):
                matches.append(f"{lineno}: {line.strip()}")
                break  # 匹配到一个模式就停止检测该行

# 保存匹配结果
with open(output_file, "w", encoding="utf-8") as f:
    for m in matches:
        f.write(m + "\n")

print(f"共匹配到 {len(matches)} 行，已保存到 {output_file}")

# flag{451}

运行得到flag为：flag{451}

JWT_Weak_Secret

import jwt
import json
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives import serialization

# 读取文件
with open('./wordlist.txt', 'r') as f:
    wordlist = [line.strip() for line in f]

with open('./tokens.txt', 'r') as f:
    tokens = [line.strip() for line in f]

with open('./public.pem', 'r') as f:
    public_key = f.read()

# 定义验证函数
def verify_jwt(token):
    # 获取JWT头部来确定算法类型
    try:
        header = json.loads(jwt.utils.base64url_decode(token.split('.')[0]).decode('utf-8'))
        algorithm = header['alg']
    except:
        return False, None

    try:
        # 如果是RS256算法，使用公钥验证
        if algorithm == 'RS256':
            payload = jwt.decode(token, public_key, algorithms=['RS256'])
            return True, payload
        # 如果是HS256算法，尝试wordlist中的每个密码
        elif algorithm == 'HS256':
            for password in wordlist:
                try:
                    payload = jwt.decode(token, password, algorithms=['HS256'])
                    return True, payload
                except (jwt.exceptions.InvalidSignatureError, jwt.exceptions.DecodeError):
                    continue
    except Exception as e:
        pass

    return False, None

# 检查是否有管理员权限
def has_admin_privileges(payload):
    # 检查admin=true
    if payload.get('admin') == True:
        return True

    # 检查role∈{admin, superuser}
    role = payload.get('role')
    if role in ['admin', 'superuser']:
        return True

    return False

# 验证所有令牌并找出符合条件的令牌
valid_admin_tokens = []

for i, token in enumerate(tokens):
    is_valid, payload = verify_jwt(token)

    if is_valid and payload:
        if has_admin_privileges(payload):
            valid_admin_tokens.append(i+1)  # 令牌序号从1开始

# 按序号排序
valid_admin_tokens.sort()

# 生成flag格式
flag = 'flag{' + ':'.join(map(str, valid_admin_tokens)) + '}'
print(flag)

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

运行得到flag为：flag{1:3:4:5:9:14:15:17:19:20:21:24:27:28}

DB_Log

import hashlib
import os

# 获取当前脚本所在目录
current_dir = os.path.dirname(os.path.abspath(__file__))

# 部门数据表分布
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']
}

# 敏感字段
sensitive_fields = ['salary', 'ssn', 'phone', 'email', 'address']

def read_user_permissions():
    """
    读取用户权限信息
    返回: (user_departments, user_allowed_tables, admins)
    """
    user_departments = {}  # 用户所属部门
    user_allowed_tables = {}  # 用户被授权访问的表
    admins = set()  # 管理员用户集合
    
    file_path = os.path.join(current_dir, 'user_permissions.txt')
    
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            for line in f:
                parts = line.strip().split(', ')
                if len(parts) >= 6:
                    user_id = parts[1]
                    department = parts[2]
                    tables = parts[3].split(';')
                    role = parts[5]
                    
                    user_departments[user_id] = department
                    user_allowed_tables[user_id] = tables
                    if role == 'admin':
                        admins.add(user_id)
    except FileNotFoundError:
        print(f"错误：找不到 {file_path} 文件")
    except Exception as e:
        print(f"读取用户权限文件时出错: {e}")
    
    return user_departments, user_allowed_tables, admins

def get_table_department(table_name):
    """
    根据表名获取所属部门
    """
    for department, tables in department_tables.items():
        if table_name in tables:
            return department
    return None

def detect_violations():
    """
    检测所有类型的违规行为
    """
    violations = []
    
    # 获取用户权限信息
    user_departments, user_allowed_tables, admins = read_user_permissions()
    if not user_departments:
        print("无法获取用户权限信息")
        return violations
    
    # 构建日志文件路径
    log_file_path = os.path.join(current_dir, 'database_logs.txt')
    
    try:
        with open(log_file_path, 'r', encoding='utf-8') as f:
            for line in f:
                # 解析日志行
                parts = line.strip().split(' ')
                if len(parts) < 4:
                    continue
                
                log_id = parts[0]
                user_id = parts[3]
                
                # 解析时间以检测规则3（工作时间外操作）
                try:
                    hour = int(parts[2].split(':')[0])
                except:
                    hour = -1  # 无法解析时间
                
                # 规则3：工作时间外操作异常（凌晨0-5点）
                if 0 <= hour < 5:
                    violations.append((3, log_id))
                
                # 其他规则需要更多日志信息
                if len(parts) >= 5:
                    operation = parts[4]
                    
                    # 规则4：数据备份异常操作（非管理员执行BACKUP）
                    if operation == 'BACKUP' and user_id in user_departments:
                        if user_id not in admins:
                            violations.append((4, log_id))
                    
                    # 需要表名的操作
                    if len(parts) >= 6 and operation == 'QUERY':
                        table_name = parts[5]
                        
                        # 规则1：跨部门数据访问违规
                        if user_id in user_departments:
                            user_department = user_departments[user_id]
                            table_department = get_table_department(table_name)
                            
                            # 检查表是否属于某个部门且用户访问了非本部门的表
                            if table_department and user_department != table_department:
                                violations.append((1, log_id))
                        
                        # 规则2：敏感字段访问违规
                        # 检查字段参数
                        for part in parts:
                            if part.startswith('field='):
                                field_value = part.split('=')[1]
                                if field_value in sensitive_fields:
                                    violations.append((2, log_id))
                                    break
    
    except FileNotFoundError:
        print(f"错误：找不到 {log_file_path} 文件")
    except Exception as e:
        print(f"处理日志文件时出错: {e}")
    
    return violations

def main():
    print("开始分析数据库日志...")
    violations = detect_violations()
    
    if not violations:
        print("未发现违规行为")
        return
    
    # 去重并按照日志ID排序
    unique_violations = list(set(violations))
    
    # 按照日志ID排序，如果日志ID相同则按规则编号排序
    unique_violations.sort(key=lambda x: (int(x[1]), x[0]))
    
    # 构建违规记录字符串
    violation_strings = [f"{rule}-{log_id}" for rule, log_id in unique_violations]
    violation_record = ",".join(violation_strings)
    
    print(f"违规记录: {violation_record}")
    
    # 计算MD5值
    md5_hash = hashlib.md5(violation_record.encode()).hexdigest()
    print(f"flag{{{md5_hash}}}")

if __name__ == "__main__":
    main()

# flag{1ff4054d20e07b42411bded1d6d895cf}

运行得到flag为：flag{1ff4054d20e07b42411bded1d6d895cf}

AES_Custom_Padding

import base64
from Crypto.Cipher import AES


def decrypt_aes_cbc_custom_padding(ciphertext_b64, key_hex, iv_hex):
    # 解码Base64密文
    ciphertext = base64.b64decode(ciphertext_b64)

    # 转换十六进制密钥和IV为字节
    key = bytes.fromhex(key_hex)
    iv = bytes.fromhex(iv_hex)

    # 创建AES-CBC解密器
    cipher = AES.new(key, AES.MODE_CBC, iv)

    # 解密
    padded_plaintext = cipher.decrypt(ciphertext)

    # 去除自定义填充
    # 查找最后一个0x80字节
    last_80_pos = padded_plaintext.rfind(b'\x80')
    if last_80_pos == -1:
        raise ValueError("无效填充 - 没有找到0x80标记")

    # 检查填充区域是否全是0x00
    padding_area = padded_plaintext[last_80_pos + 1:]
    if any(byte != 0x00 for byte in padding_area):
        raise ValueError("无效填充 - 填充区域包含非零字节")

    # 提取真实明文
    plaintext = padded_plaintext[:last_80_pos]

    return plaintext


# 给定的参数
ciphertext_b64 = "WU+8dpscYYw+q52uQqX8OPiesnTajq++AXj05zX3u9an27JZR9/31yZaWdtPM5df"
key_hex = "0123456789ABCDEF0123456789ABCDEF"
iv_hex = "000102030405060708090A0B0C0D0E0F"

# 解密
try:
    plaintext = decrypt_aes_cbc_custom_padding(ciphertext_b64, key_hex, iv_hex)
    print("解密成功!")
    print("明文:", plaintext)
    print("明文(十六进制):", plaintext.hex())
except Exception as e:
    print("解密失败:", str(e))

# 解密成功!
# 明文: b'flag{T1s_4ll_4b0ut_AES_custom_padding!}'
# 明文(十六进制): 666c61677b5431735f346c6c5f34623075745f4145535f637573746f6d5f70616464696e67217d

运行得到flag为：flag{T1s_4ll_4b0ut_AES_custom_padding!}

Brute_Force_Detection

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


def parse_log_line(line):
    """解析日志行，提取时间、结果、用户名和IP地址"""
    # 适配日志格式："YYYY-mm-dd HH:MM:SS RESULT user=<user> ip=<ip>"
    pattern = r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) (\w+) user=(\w+) ip=(\d+\.\d+\.\d+\.\d+)'
    match = re.match(pattern, line.strip())

    if match:
        timestamp_str, result, user, ip = match.groups()
        timestamp = datetime.strptime(timestamp_str, '%Y-%m-%d %H:%M:%S')
        return {
            'timestamp': timestamp,
            'result': result,
            'user': user,
            'ip': ip
        }
    return None


def detect_brute_force(log_file):
    """检测符合条件的IP地址"""
    # 按IP和用户分组存储日志记录
    ip_user_logs = defaultdict(list)

    # 读取并解析日志文件
    with open(log_file, 'r', encoding='utf-8') as f:  # 增加编码参数，避免中文乱码问题
        for line in f:
            log_entry = parse_log_line(line)
            if log_entry:
                key = (log_entry['ip'], log_entry['user'])
                ip_user_logs[key].append(log_entry)

    # 存储符合条件的IP
    suspicious_ips = set()

    # 检查每个IP对每个用户的登录记录
    for (ip, user), entries in ip_user_logs.items():
        # 按时间排序
        entries.sort(key=lambda x: x['timestamp'])

        # 检查连续失败记录
        fail_sequence = []
        for entry in entries:
            if entry['result'] == 'FAIL':
                fail_sequence.append(entry)
                # 只保留最近的5次失败记录
                if len(fail_sequence) > 5:
                    fail_sequence.pop(0)
            else:  # SUCCESS
                # 检查是否前面刚好有5次失败
                if len(fail_sequence) == 5:
                    # 检查5次失败是否在10分钟内
                    time_diff = fail_sequence[-1]['timestamp'] - fail_sequence[0]['timestamp']
                    if time_diff <= timedelta(minutes=10):
                        suspicious_ips.add(ip)
                # 重置失败序列
                fail_sequence = []

    # 按IP地址排序（按数字大小）
    sorted_ips = sorted(suspicious_ips, key=lambda x: [int(part) for part in x.split('.')])

    # 生成flag格式
    return f"flag{{{':'.join(sorted_ips)}}}"


if __name__ == "__main__":
    log_file_path = "./auth.log"
    result = detect_brute_force(log_file_path)
    print(result)

# flag{192.168.3.13:192.168.5.15}

运行得到flag为：flag{192.168.3.13:192.168.5.15}