文章总结： 文档详细介绍了2026年第十届御网杯数据安全赛道中TracePurge和ShadowMeter两个题目的解题思路与实现过程。TracePurge部分通过分析CSV文件匹配客户信息生成数据泄露响应计划，ShadowMeter部分则利用access.log和error.log恢复被分片外传的ZIP文件，展示了数据泄露应急响应和日志分析的关键技术。 综合评分： 85 文章分类： CTF,数据安全,应急响应,安全工具
赛查查
2026年8月4日%2012:23 北京
在小说阅读器读本章
去阅读
以下文章来源于云晞科技Sec ，作者江思澄
云晞科技Sec .
这里是网络安全探索基地！专注分享流量分析实战技巧、应急响应解决方案，深度剖析Web安全漏洞攻防。同时带来CTF竞赛解题思路与精彩笔记，助你快速掌握前沿技术，提升安全技能，共同筑牢数字世界安全防线！
解压后先看了事件响应说明.md
这里最重要的规则就三件事：%20样本用%20email_hash%20或%20name%20+%20phone_mask%20匹配客户；只处理本次泄露对应的导出批次；最后生成%20leak_id,subject_id,export_id,operator,action
然后看了这四个%20CSV%20的表头和前几行
import csv
from collections import Counter
base%20= r'D:\下载\8：TracePurge的附件\tempdir\DS附件\TracePurge附件\TracePurge附件'
leak_path%20=%20base%20+ r'\leak_sample.csv'
cnt%20=%20Counter()
with open(leak_path,%20newline='',%20encoding='utf-8-sig') as f:
%20 for row in csv.DictReader(f):
%20 %20 %20 %20cnt[row['batch_tag']]%20+= 1
for k,%20v in cnt.most_common():
%20 %20print(k,%20v)
CRM-20260508-D4%20数量明显最多
所以这次事件就是%20EX20260508004，操作人是%20ops_chen
后面按说明匹配客户%20email%20非空就%20strip%20+%20lower%20后做%20sha256，去%20customer_index.csv%20的%20email_hash%20里找邮箱为空或者哈希找不到，就用%20name%20+%20phone_mask，只有唯一命中才算一个%20subject_id%20如果出现多次，保留%20leak_id%20字典序最小的那条
直接生成提交文件：
import csv
import hashlib
from collections import defaultdict,%20Counter
base%20= r'D:\下载\8：TracePurge的附件\tempdir\DS附件\TracePurge附件\TracePurge附件'
out_path%20= r'D:\下载\8：TracePurge的附件\tempdir\DS附件\TracePurge附件\TracePurge附件\response_plan.csv'
leak_path%20=%20base%20+ r'\leak_sample.csv'
idx_path%20=%20base%20+ r'\customer_index.csv'
target_batch%20= 'CRM-20260508-D4'
export_id%20= 'EX20260508004'
operator%20= 'ops_chen'
email_map%20=%20{}
name_phone%20=%20defaultdict(list)
with open(idx_path,%20newline='',%20encoding='utf-8-sig') as f:
%20 for row in csv.DictReader(f):
%20 %20 %20 if row['email_hash']:
%20 %20 %20 %20 %20 %20email_map[row['email_hash']]%20=%20row
%20 %20 %20 %20name_phone[(row['name'],%20row['phone_mask'])].append(row)
def match_subject(leak):
%20 %20email%20=%20leak['email'].strip()
%20 if email:
%20 %20 %20 %20h%20=%20hashlib.sha256(email.lower().encode()).hexdigest()
%20 %20 %20 if h in email_map:
%20 %20 %20 %20 %20 return email_map[h]
%20 %20hits%20=%20name_phone[(leak['name'],%20leak['phone_mask'])]
%20 if len(hits)%20== 1:
%20 %20 %20 return hits[0]
%20 returnNone
def get_action(subject):
%20 if subject['erase_requested']%20== 'true':
%20 %20 %20 return'PURGE'
%20 if subject['risk']%20== 'high':
%20 %20 %20 return'NOTIFY'
%20 return'MONITOR'
best%20=%20{}
with open(leak_path,%20newline='',%20encoding='utf-8-sig') as f:
%20 for leak in csv.DictReader(f):
%20 %20 %20 if leak['batch_tag']%20!=%20target_batch:
%20 %20 %20 %20 %20 continue
%20 %20 %20 %20subject%20=%20match_subject(leak)
%20 %20 %20 ifnot subject:
%20 %20 %20 %20 %20 continue
%20 %20 %20 %20sid%20=%20subject['subject_id']
%20 %20 %20 %20row%20=%20{
%20 %20 %20 %20 %20 'leak_id':%20leak['leak_id'],
%20 %20 %20 %20 %20 'subject_id':%20sid,
%20 %20 %20 %20 %20 'export_id':%20export_id,
%20 %20 %20 %20 %20 'operator':%20operator,
%20 %20 %20 %20 %20 'action':%20get_action(subject),
%20 %20 %20 %20}
%20 %20 %20 if sid notin best or leak['leak_id']%20<%20best[sid]['leak_id']:
%20 %20 %20 %20 %20 %20best[sid]%20=%20row
rows%20=%20sorted(best.values(),%20key=lambda r:%20(r['leak_id'],%20r['subject_id']))
with open(out_path, 'w',%20newline='',%20encoding='utf-8') as f:
%20 %20writer%20=%20csv.DictWriter(
%20 %20 %20 %20f,
%20 %20 %20 %20fieldnames=['leak_id', 'subject_id', 'export_id', 'operator', 'action']
%20 %20)
%20 %20writer.writeheader()
%20 %20writer.writerows(rows)
print('output:',%20out_path)
print('rows:',%20len(rows))
print(Counter(r['action'] for r in rows))
生成出来%20response_plan.csv%20一共%201200%20行，动作分布是%20MONITOR%20964、NOTIFY%20147、PURGE%2089上传平台后%20100%
flag{e115cc3382d6b63c3196a4f1ca32eefa}
这题给了%20access.log%20和%20error.log我先看%20access.log，登录接口这里没有一行行数，直接把状态码统计出来：
结果是%207%20次%20401、1%20次%20200，说明确实有一次登录成功access.log%20后面还能看到%20/admin/export.php和/collect/report.php%20开始大量出现，所以我怀疑是先登录导出，再通过%20collect%20接口分片外传
access.log%20看不到%20POST%20body，所以继续翻%20error.log这里%20Apache%20开了%20dumpio，会把请求体和响应也记下来我先搜%20/admin/login.php，找到最后一次登录请求的%20client%20是%20172.25.0.1:60636，再用这个%20client%20过滤同一次请求：
这时候才看到登录的%20body%20和响应：
POST%20/admin/login.php%20HTTP/1.1
username=admin_ops&password=Admin%40Pa%24%24w0rd
{"token":"ops-internal-8dd18cfd"}
Admin%40Pa%24%24w0rd%20URL%20解码后是：
Admin@Pa$$w0rd
登录这块确认完以后，继续看%20/collect/report.php%20的%20POST%20bodyerror.log%20里能看到每次上报都有%20job、seq、notejob=SM-20260507-17%20里数字%20seq%20的%20note%20是%20base64%20分片，第一片以%20UEsDB%20开头，base64%20解出来是%20PK\x03\x04，说明是在分片外传%20zip
再看结束分片，seq=END%20的%20note%20不是%20base64，而是给了校验信息：
所以恢复思路就是按%20seq%20顺序拼接数字分片的%20note，中间重复%20seq%20用第一份，最后用%20archive_size=10601%20和%20sha256=4019ef293eeb8af8a0d5c0e66c0bdb3cbcd17067839f42c86beb501a83d6c59b%20校验，能对上就说明%20zip%20恢复对了
import base64,%20codecs,%20collections,%20datetime,%20hashlib,%20pathlib,%20re,%20urllib.parse
err%20=%20pathlib.Path(r"D:\下载\3：ShadowMeter的附件\tempdir\MISC附件\ShadowMeter附件\ShadowMeter附件\error.log")
out%20=%20pathlib.Path(r"D:\下载\3：ShadowMeter的附件\tempdir\MISC附件\ShadowMeter附件\ShadowMeter附件\1.zip")
line_re%20=%20re.compile(
%20 r"^\[(?P<ts>[^\]]+)\].*?\[client%20(?P<client>[^\]]+)\]%20"
%20 r"mod_dumpio:\s+dumpio_(?P<dir>in|out)%20\((?P<kind>[^)]+)\):%20(?P<data>.*)$"
)
def ts(s):
%20 return datetime.datetime.strptime(s, "%a%20%b%20%d%20%H:%M:%S.%f%20%Y")
def unesc(s):
%20 return codecs.decode(s.encode(), "unicode_escape").encode("latin-1", "replace")
streams%20=%20collections.defaultdict(list)
for line in err.read_text(errors="replace").splitlines():
%20 %20m%20=%20line_re.match(line)
%20 ifnot m:
%20 %20 %20 continue
%20 if m.group("dir")%20!= "in":
%20 %20 %20 continue
%20 ifnot m.group("kind").startswith("data-"):
%20 %20 %20 continue
%20 %20data%20=%20m.group("data")
%20 if re.fullmatch(r"\d+%20bytes",%20data):
%20 %20 %20 continue
%20 %20streams[m.group("client")].append((ts(m.group("ts")),%20unesc(data)))
def parse_requests(buf):
%20 %20pos%20= 0
%20 while pos%20<%20len(buf):
%20 %20 %20 %20end%20=%20buf.find(b"\r\n\r\n",%20pos)
%20 %20 %20 if end%20< 0:
%20 %20 %20 %20 %20 break
%20 %20 %20 %20head%20=%20buf[pos:end]
%20 %20 %20 %20first%20=%20head.split(b"\r\n", 1)[0].decode("latin-1", "replace")
%20 %20 %20 %20m%20=%20re.match(r"(GET|POST|OPTIONS)%20([^%20]+)%20HTTP/",%20first)
%20 %20 %20 ifnot m:
%20 %20 %20 %20 %20 %20nxt%20=%20buf.find(b"POST%20",%20pos%20+ 1)
%20 %20 %20 %20 %20 if nxt%20< 0:
%20 %20 %20 %20 %20 %20 %20 break
%20 %20 %20 %20 %20 %20pos%20=%20nxt
%20 %20 %20 %20 %20 continue
%20 %20 %20 %20headers%20=%20{}
%20 %20 %20 for line in head.split(b"\r\n")[1:]:
%20 %20 %20 %20 %20 ifb":"in line:
%20 %20 %20 %20 %20 %20 %20 %20k,%20v%20=%20line.split(b":", 1)
%20 %20 %20 %20 %20 %20 %20 %20headers[k.decode().lower()]%20=%20v.strip().decode()
%20 %20 %20 %20clen%20=%20int(headers.get("content-length", "0") or0)
%20 %20 %20 %20body%20=%20buf[end%20+ 4:end%20+ 4 +%20clen]
%20 %20 %20 yield m.group(1),%20m.group(2),%20body
%20 %20 %20 %20pos%20=%20end%20+ 4 +%20clen
rows%20=%20[]
for parts in streams.values():
%20 %20buf%20= b"".join(x[1] for x in sorted(parts))
%20 for method,%20path,%20body in parse_requests(buf):
%20 %20 %20 if path.split("?")[0]%20!= "/collect/report.php":
%20 %20 %20 %20 %20 continue
%20 %20 %20 %20qs%20=%20urllib.parse.parse_qs(body.decode("latin-1"),%20keep_blank_values=True)
%20 %20 %20 %20rows.append({k:%20v[0] for k,%20v in qs.items()})
sm%20=%20[r for r in rows if r.get("job")%20== "SM-20260507-17"]
end_note%20=%20next(r["note"] for r in sm if r.get("seq")%20== "END")
target_size%20=%20int(re.search(r"archive_size=(\d+)",%20end_note).group(1))
target_sha%20=%20re.search(r"sha256=([0-9a-f]+)",%20end_note).group(1)
chunks%20=%20{}
for r in sm:
%20 %20seq%20=%20r.get("seq", "")
%20 if seq.isdigit():
%20 %20 %20 %20chunks.setdefault(int(seq),%20r["note"])
b64%20= "".join(chunks[i] for i in range(1,%20max(chunks)%20+ 1))
data%20=%20base64.b64decode(b64)
print(len(data))
print(hashlib.sha256(data).hexdigest())
assert len(data)%20==%20target_size
assert hashlib.sha256(data).hexdigest()%20==%20target_sha
out.write_bytes(data)
print(out)
zip%20需要密码，我本来拿登录密码%20Admin@Pa$$w0rd%20试了，不对继续搜%20debug%20相关请求：
grep%20-nE 'POST%20/admin/debug\.php|peek_export' error.log
结果两行都是同一个%20client%20端口%20172.25.0.1:55612，说明请求体属于这次%20/admin/debug.php：
POST%20/admin/debug.php%20HTTP/1.1
op=peek_export_keyf&token=meter_report&key=SM-20260507-17%3Ameter%3Areport
把%20SM-20260507-17%3Ameter%3Areport%20URL%20解码，得到%20zip%20密码：
SM-20260507-17:meter:report
拿这个密码解压%201.zip：
解出来以后有%20export.csv、manifest.json、readme.txtmanifest.json%20里%20focus.filter%20写的是%20app=portal、metric_name=secure.checkpoint、user_id=ADMIN_OPS，所以去%20export.csv%20里找这条目标记录最后命中：
flag{c1b50d61ac8b81fac5ab3ea499056c0a}
这题给了个双层%20zip，外层先正常解，里面是%20backup_2025Q1.zip，发现有密码尝试了一些弱口令，尝试了一下文件名%20202501解压成功
解出来的%20png%20一开始打不开，头部前%204%20字节是%2000%2000%2000%2000
补回%20PNG%20magic%2089%2050%204e%2047%20后能识别成%20800×400
但%20pngcheck%20报%20IHDR%20CRC%20错这里卡了一下，本来以为就是%20CRC%20坏了，后来拿原%20CRC%205412913f%20去爆宽高，发现它正好对应%20800×800，说明高度被人从%20800%20改成了%20400，下半张图被藏起来了把高度改回%20800%20后图片校验通过，下半部分直接写了敏感信息和%20flag
flag{b2607454056ecb4d3dc9375a6fb76fdb}
前%208%20字节大概是：
51%207e%207f%20ad%2015%2035%207c%20a9
然后想一下，.docx%20本质是%20zip，zip%20文件头通常是：
50%204b%2003%2004%2014%2000%2000%2000
所以密钥可以用密文%20XOR%20明文头推出：
c%20=%20bytes.fromhex("51%207e%207f%20ad%2015%2035%207c%20a9")%20p%20=%20bytes.fromhex("50%204b%2003%2004%2014%2000%2000%2000")%20print(bytes(a%20^%20b for a,%20b in zip(c,%20p)).hex())
说明%20key%20是循环的%2001%2035%207c%20a9
解密exp
from pathlib import Path
import zipfile
enc%20=%20Path(r'.\task.docx.enc')
out%20=%20Path(r'.\task.docx')
key%20=%20bytes.fromhex('01357ca9')
data%20=%20enc.read_bytes()
plain%20=%20bytes(b%20^%20key[i%20%%20len(key)] for i,%20b in enumerate(data))
out.write_bytes(plain)
print(plain[:16].hex())
print(zipfile.is_zipfile(out))
解出来以后%20plain[:16]%20是%20504b03041400000008008bb8a75cad52，zipfile.is_zipfile%20返回%20True，说明%20docx%20复原了
这个看着不像最终%20flag，但%20synt%20ROT13%20后是%20flag
flag{ad213351-9510-46c0-8622-9cdea9f634d2}
先看%20个人信息数据规范文档.md%20正要检查的规则就这些：
username：只能大小写字母、数字、下划线
name：只能中文
phone：11%20位数字，前三位必须在指定号段里
email：必须有%20@，@%20前后有内容，并且有域名后缀
gender：只能是%20男%20或%20女
打开%20user_data.db
insurance_users，字段是：
id,%20username,%20name,%20phone,%20email,%20gender
发现有好多错误的数据，要按照个人信息数据规范文档.md%20开始清理一下
例：
name%20为空，不是中文姓名
email%20是%[email protected]，@%20前面没有内容
gender%20是%20b，不是%20男/女
平台样例文件也要下，看它到底要什么格式
sql筛选exp:
SELECT%20id,%20username,%20name,%20phone,%20email,%20gender
FROM%20insurance_users
WHERE
%20 %20username%20GLOB%20'*[^A-Za-z0-9_]*'
%20 %20OR%20name%20=%20''
%20 %20OR%20name%20GLOB%20'*[^一-龥]*'
%20 %20OR%20length(phone)%20!=%2011
%20 %20OR%20phone%20GLOB%20'*[^0-9]*'
%20 %20OR%20substr(phone,%201,%203)%20NOT%20IN%20(
%20 %20 %20 %20'134','135','136','137','138','139','147','148','150','151','152','157','158','159',
%20 %20 %20 %20'172','178','182','183','184','187','188','195','198','130','131','132','140','145',
%20 %20 %20 %20'146','155','156','166','167','171','175','176','185','186','196','133','149','153',
%20 %20 %20 %20'173','174','177','180','181','189','190','191','193','199'
%20 %20)
%20 %20OR%20length(email)%20-%20length(replace(email,%20'@',%20''))%20!=%201
%20 %20OR%20email%20NOT%20GLOB%20'?*@?*.?*'
%20 %20OR%20gender%20NOT%20IN%20('男',%20'女')
ORDER%20BY%20id;
龥%20是一个汉字，常被拿来当“常用%20CJK%20汉字范围”的右边界
导出上传%20csv%20文件上传校验%20获得%20flag%20值
flag{9d4c2f88e16b70a5c39f1e27bd84c610}
核心就是先筛%20status=active，再按%20id%20升序导出固定表头%20CSV各字段基本都是先校验再脱敏，不合法写%20INVALID用户名、姓名、邮箱用户名都是保留首尾打星%20密码做%20MD5；身份证要校验日期和校验位后只留年份%20手机号和银行卡先归一化数字再判断长度/号段%20地址必须有“号”%20IP%20要合法%20IPv4；生日按%202026-05-01%20算十年年龄段
随机创个%20db%20文件%20，导入%20user_export.sql%20文件查看一下
这里能看到%20SQL%20是单表%20user_export，字段就是：
id,%20username,%20password,%20name,%20idcard,%20phone,%20email,%20bankcard,%20address,%20ip,%20birthday,%20status
看下示例文件
这一步很重要，示例前几行可以用来确认自己脱敏规则对不对
我先在%20DB%20Browser%20的执行%20SQL%20里查了一下数据，只筛%20status=active并按%20id%20升序：
SELECT%20id,%20username,%20password,%20name,%20idcard,%20phone,%20email,%20bankcard,%20address,%20ip,%20birthday
FROM%20user_export
WHERE%20status%20= 'active'
ORDER%20BY%20CAST(id%20AS%20INTEGER)%20ASC;
查出来%20active%20一共%207000%20条，说明筛选范围没问题后面的脱敏规则比较多，比如密码要%20MD5，身份证要校验出生日期和最后一位校验码，手机号还要处理%20+86、空格、横杠这些格式，所以我没有继续硬写%20SQL，直接用%20Python%20从%20SQLite%20里读%20active%20数据，再逐字段处理生成%20CSV
exp:
import csv,%20re,%20hashlib,%20sqlite3,%20ipaddress
from datetime import datetime,%20date
from pathlib import Path
base%20=%20Path(__file__).resolve().parent
sql%20=%20next(base.rglob("user_export.sql"))
out%20=%20base%20/ "example.csv"
phones%20=%20set("""134%20135%20136%20137%20138%20139%20147%20148%20150%20151%20152%20157%20158%20159%20172%20178%20182%20183%20184%20187%20188%20195%20198
130%20131%20132%20140%20145%20146%20155%20156%20166%20167%20171%20175%20176%20185%20186%20196
133%20149%20153%20173%20174%20177%20180%20181%20189%20190%20191%20193%20199""".split())
weights%20=%20[7, 9, 10, 5, 8, 4, 2, 1, 6, 3, 7, 9, 10, 5, 8, 4, 2]
checks%20= "10X98765432"
email_re%20=%20re.compile(r"^[a-z0-9._%+-]+@[a-z0-9.-]+\.[a-z]{2,}$")
def star(s):
%20 return s[0]%20+ "*"if len(s)%20== 2else s[0]%20+ "*" *%20(len(s)%20- 2)%20+%20s[-1]
def name(s):
%20 return star(s) if len(s)%20>= 2and re.fullmatch(r"[\u4e00-\u9fff]+",%20s) else"INVALID"
def idcard(s):
%20 %20s%20=%20s.upper()
%20 ifnot re.fullmatch(r"\d{17}[\dX]",%20s):
%20 %20 %20 return"INVALID"
%20 try:
%20 %20 %20 %20datetime.strptime(s[6:14], "%Y%m%d")
%20 except ValueError:
%20 %20 %20 return"INVALID"
%20 if checks[sum(int(a)%20*%20b for a,%20b in zip(s[:17],%20weights))%20% 11]%20!=%20s[-1]:
%20 %20 %20 return"INVALID"
%20 return"******" +%20s[6:10]%20+ "********"
def phone(s):
%20 %20d%20=%20re.sub(r"\D", "",%20s)
%20 if len(d)%20== 13and d.startswith("86"):
%20 %20 %20 %20d%20=%20d[2:]
%20 return d[:3]%20+ "****" +%20d[7:] if len(d)%20== 11and d[:3] in phones else"INVALID"
def email(s):
%20 %20s%20=%20s.lower()
%20 ifnot email_re.fullmatch(s):
%20 %20 %20 return"INVALID"
%20 %20a,%20b%20=%20s.split("@", 1)
%20 return star(a)%20+ "@" +%20b
def bank(s):
%20 %20d%20=%20re.sub(r"\D", "",%20s)
%20 return d[:6]%20+ "*" *%20(len(d)%20- 10)%20+%20d[-4:] if re.fullmatch(r"\d{16,19}",%20d) else"INVALID"
def addr(s):
%20 %20i%20=%20s.find("号")
%20 return s[:i]%20+ "***"if i%20!= -1else"INVALID"
def ip(s):
%20 try:
%20 %20 %20 %20p%20=%20str(ipaddress.IPv4Address(s)).split(".")
%20 %20 %20 return".".join(p[:3]%20+%20["*"])
%20 except:
%20 %20 %20 return"INVALID"
def birthday(s):
%20 %20fmt%20= "%Y-%m-%d"if"-"in s else"%Y/%m/%d"if"/"in s elseNone
%20 ifnot fmt:
%20 %20 %20 return"INVALID"
%20 try:
%20 %20 %20 %20b%20=%20datetime.strptime(s,%20fmt).date()
%20 except:
%20 %20 %20 return"INVALID"
%20 %20ref%20=%20date(2026, 5, 1)
%20 %20age%20=%20ref.year%20-%20b.year%20-%20((ref.month,%20ref.day)%20<%20(b.month,%20b.day))
%20 %20lo%20=%20age%20// 10 * 10
%20 returnf"{lo}-{lo+9}"
con%20=%20sqlite3.connect(":memory:")
con.executescript(sql.read_text(encoding="utf-8"))
rows%20=%20con.execute("""
SELECT%20id,%20username,%20password,%20name,%20idcard,%20phone,%20email,%20bankcard,%20address,%20ip,%20birthday
FROM%20user_export
WHERE%20status='active'
ORDER%20BY%20CAST(id%20AS%20INTEGER)
""").fetchall()
with out.open("w",%20encoding="utf-8",%20newline="") as f:
%20 %20w%20=%20csv.writer(f,%20lineterminator="\n")
%20 %20w.writerow("id%20username%20password%20name%20idcard%20phone%20email%20bankcard%20address%20ip%20birthday".split())
%20 for r in rows:
%20 %20 %20 %20i,%20u,%20p,%20n,%20c,%20ph,%20e,%20bk,%20ad,%20ipaddr,%20bd%20=%20r
%20 %20 %20 %20w.writerow([
%20 %20 %20 %20 %20 %20i,
%20 %20 %20 %20 %20 %20star(u),
%20 %20 %20 %20 %20 %20hashlib.md5(p.encode()).hexdigest(),
%20 %20 %20 %20 %20 %20name(n),
%20 %20 %20 %20 %20 %20idcard(c),
%20 %20 %20 %20 %20 %20phone(ph),
%20 %20 %20 %20 %20 %20email(e),
%20 %20 %20 %20 %20 %20bank(bk),
%20 %20 %20 %20 %20 %20addr(ad),
%20 %20 %20 %20 %20 %20ip(ipaddr),
%20 %20 %20 %20 %20 %20birthday(bd)
%20 %20 %20 %20])
print(out)
print("active%20rows:",%20len(rows))
上传csv文件%20校验得到flag
flag{d11d22f3398ef4748b21ef061793612f}
附件解出来以后先看文件类型和%20hash，能看到%20exfil_tool.pyc%20是%20CPython%203.11%20的%20pyc，notice.docx%20是%20Word%20文档，poster.png 是 512×512 PNG
先从 pyc 下手，反汇编看一下里面有什么：
python -m xdis.bin.pydisasm exfil_tool.pyc
a1b2c3d4e5f60718
最后把二维码结果拼上%20docx%20结果当%20key：
a1b2c3d4e5f60718c9d0e1f2a3b4c5d6
pyc%20里的%20_0x4f3a%20是左旋再%20xor，反解的时候就是先%20xor%20key，再按%20i%20%%208%20右旋：
cipher%20=%20bytes.fromhex(
%20 "c76a46ef52b009aafebc2553a0939c4e95de0b1dd6509f00f81a74638093cb6698d44a758359"
)
key%20=%20bytes.fromhex("a1b2c3d4e5f60718c9d0e1f2a3b4c5d6")
def ror(x,%20s):
%20 return ((x%20>>%20s)%20|%20(x%20<<%20(8 -%20s)))%20& 0xff if s else x
flag%20=%20bytes(
%20 %20ror(cipher[i]%20^%20key[i%20%%20len(key)],%20i%20% 8)
%20 for i in range(len(cipher))
)
print("key%20=",%20key.hex())
print("flag%20=",%20flag.decode())
最后拿到：
flag{28e761409e1462935b01ee2298a93b4f}
用火眼取证分析打开附件解压出来的img%20在这些路径下发现了被删除的痕迹
/home/audit/.cache/sectorvault/
删掉的扫描缓存和发布包都还在镜像里，直接恢复，右键导出
查看%20/home/audit/Documents/数据分级与敏感信息识别规范.md
重点是：
仅统计%20risk_level%20>=%203%20的记录
同一%20category%20与%20normalized_value%20重复时只保留%20source_path%20字典序最小、line_no%20最小的一条
release_policy:%20ready_record_with_matching_count_and_sha256
password_policy:%20SV-{case_id}-{effective_total}-P{phone_count}I{idcard_count}B{bankcard_count}E{email_count}-{evidence_digest8}
用%20DB%20Browser%20打开%20scan_cache.sqlite
这里就能证明要开的不是%20old，也不是%20draft，而是%20ready%20的%20audit_release_20260507.zip
有效记录和口令直接用脚本
import sqlite3,%20hashlib,%20datetime,%20re
db%20= r"D:\下载\123\scan_cache.sqlite"
phone_prefixes%20=%20set("""134%20135%20136%20137%20138%20139%20147%20148%20150%20151%20152%20157%20158%20159%20172%20178%20182%20183%20184%20187%20188%20195%20198%20130%20131%20132%20140%20145%20146%20155%20156%20166%20167%20171%20175%20176%20185%20186%20196%20133%20149%20153%20173%20174%20177%20180%20181%20189%20190%20191%20193%20199""".split())
bins%20=%20set("""622848%20622700%20621700%20622262%20622188%20622200%20622568%20622609%20622908%20622518""".split())
def valid_phone(s):
%20 return bool(re.fullmatch(r"\d{11}",%20s)) and s[:3] in phone_prefixes
def valid_id(s):
%20 %20s%20=%20s.upper()
%20 ifnot re.fullmatch(r"\d{17}[0-9X]",%20s):
%20 %20 %20 returnFalse
%20 try:
%20 %20 %20 %20datetime.date(int(s[6:10]),%20int(s[10:12]),%20int(s[12:14]))
%20 except ValueError:
%20 %20 %20 returnFalse
%20 %20weights%20=%20[7, 9, 10, 5, 8, 4, 2, 1, 6, 3, 7, 9, 10, 5, 8, 4, 2]
%20 %20checks%20= "10X98765432"
%20 return checks[sum(int(a)%20*%20b for a,%20b in zip(s[:17],%20weights))%20% 11]%20==%20s[-1]
def valid_bank(s):
%20 ifnot re.fullmatch(r"\d{16,19}",%20s) ornot any(s.startswith(b) for b in bins):
%20 %20 %20 returnFalse
%20 %20total%20= 0
%20 %20parity%20=%20len(s)%20% 2
%20 for idx,%20ch in enumerate(s):
%20 %20 %20 %20d%20=%20int(ch)
%20 %20 %20 if idx%20% 2 ==%20parity:
%20 %20 %20 %20 %20 %20d%20*= 2
%20 %20 %20 %20 %20 if d%20> 9:
%20 %20 %20 %20 %20 %20 %20 %20d%20-= 9
%20 %20 %20 %20total%20+=%20d
%20 return total%20% 10 == 0
def valid_email(s):
%20 return"@"in s andnot s.lower().endswith("@internal.example")
validators%20=%20{
%20 "phone":%20valid_phone,
%20 "idcard":%20valid_id,
%20 "bankcard":%20valid_bank,
%20 "email":%20valid_email,
}
con%20=%20sqlite3.connect(db)
rows%20=%20con.execute("""
select%20findings.category,%20findings.normalized_value,%20files.source_path,%20findings.line_no
from%20findings%20join%20files%20on%20files.id%20=%20findings.file_id
where%20files.scope%20=%20'20260507'%20and%20findings.risk_level%20>=%203
""").fetchall()
kept%20=%20{}
for cat,%20norm,%20path,%20line in rows:
%20 if cat notin validators ornot validators[cat](norm):
%20 %20 %20 continue
%20 %20key%20=%20(cat,%20norm)
%20 %20rec%20=%20(cat,%20norm,%20path,%20line)
%20 if key notin kept or (path,%20line)%20<%20(kept[key][2],%20kept[key][3]):
%20 %20 %20 %20kept[key]%20=%20rec
final%20=%20sorted(kept.values(),%20key=lambda x:%20(x[0],%20x[1],%20x[2],%20x[3]))
evidence%20= "".join(f"{cat},{norm},{path},{line}\n"for cat,%20norm,%20path,%20line in final)
digest8%20=%20hashlib.sha256(evidence.encode()).hexdigest()[:8]
counts%20=%20{c:%20sum(1for r in final if r[0]%20==%20c) for c in ["phone", "idcard", "bankcard", "email"]}
password%20= f"SV-SV-20260507-{len(final)}-P{counts['phone']}I{counts['idcard']}B{counts['bankcard']}E{counts['email']}-{digest8}"
print("effective_total%20=",%20len(final))
print("counts%20=",%20counts)
print("digest8%20=",%20digest8)
print("password%20=",%20password)
这题先看README.pdf，规则里说得比较死：攻击类型只看user_input，而泄露的数据只从model_output%20里提；身份证、手机号、银行卡还要按题目给的规则校验，不然订单号、服务热线、工单号这种东西会混进去
我先用%20VS%20Code%20打开chat_logs.jsonl，搜Base64%20看一条明显的第%2099%20轮这里user_input%20是%20Base64%20注入，model_output%20里吐了email_b64，说明这类就是成功攻击，不是普通客服回复
把这一段email_b64%20丢到%20CyberChef，From%20Base64%20直接能解出邮箱[email protected]这个例子能证明后面脚本里必须处理%20Base64%20变形，不能只匹配明文邮箱
然后写脚本批量跑逻辑就三步：先按user_input%20判断encoding%20/%20unicode%20/%20role_play%20/%20direct，顺序按题目优先级来；再从model_output%20里提明文、Base64、空格/横杠分隔的%20PII；最后做身份证校验位、手机号号段、银行卡%20Luhn%20校验，能过校验才写进result.csv
这里我还踩了个坑，一开始role_play%20只写了Pretend%20you%20are%20和你的身份是，提交只有96.053%后来发现还有Pretend%20you’re%20…%20和你现在的身份是CyberAdmin…，这俩也得算角色扮演，补上后才完整
脚本如下：
import base64
import csv
import json
import re
from pathlib import Path
log%20=%20Path("chat_logs.jsonl")
out%20=%20Path("result.csv")
phone_prefix%20=%20set(
%20 "134%20135%20136%20137%20138%20139%20147%20148%20150%20151%20152%20157%20158%20159%20"
%20 "172%20178%20182%20183%20184%20187%20188%20195%20198%20"
%20 "130%20131%20132%20140%20145%20146%20155%20156%20166%20167%20171%20175%20176%20185%20186%20196%20"
%20 "133%20149%20153%20173%20174%20177%20180%20181%20189%20190%20191%20193%20199".split()
)
bank_bins%20=%20(
%20 "622848", "622700", "621700", "622262", "622188",
%20 "622200", "622568", "622609", "622908", "622518",
)
id_weights%20=%20[7, 9, 10, 5, 8, 4, 2, 1, 6, 3, 7, 9, 10, 5, 8, 4, 2]
id_check%20= "10X98765432"
def is_idcard(s):
%20 %20s%20=%20s.upper()
%20 ifnot re.fullmatch(
%20 %20 %20 r"[1-9]\d{5}(?:18|19|20)\d{2}(?:0[1-9]|1[0-2])"
%20 %20 %20 r"(?:0[1-9]|[12]\d|3[01])\d{3}[0-9X]",
%20 %20 %20 %20s,
%20 %20):
%20 %20 %20 returnFalse
%20 %20total%20=%20sum(int(a)%20*%20b for a,%20b in zip(s[:17],%20id_weights))
%20 return id_check[total%20% 11]%20==%20s[-1]
def is_phone(s):
%20 return bool(re.fullmatch(r"1\d{10}",%20s)) and s[:3] in phone_prefix
def is_bankcard(s):
%20 ifnot re.fullmatch(r"\d{16,19}",%20s) ornot s.startswith(bank_bins):
%20 %20 %20 returnFalse
%20 %20total%20= 0
%20 for i,%20ch in enumerate(s[::-1], 1):
%20 %20 %20 %20d%20=%20int(ch)
%20 %20 %20 if i%20% 2 == 0:
%20 %20 %20 %20 %20 %20d%20*= 2
%20 %20 %20 %20 %20 if d%20> 9:
%20 %20 %20 %20 %20 %20 %20 %20d%20-= 9
%20 %20 %20 %20total%20+=%20d
%20 return total%20% 10 == 0
def is_email(s):
%20 return bool(re.fullmatch(r"[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}",%20s))
role_re%20=%20re.compile(
%20 r"(假设你是|你现在是|你现在的身份是|你的身份是|你的角色是|扮演|从现在开始你是|"
%20 r"Pretend%20you%20are|Pretend%20you're|Act%20as|You%20are%20now|Let's%20play)",
%20 %20re.I,
)
direct_re%20=%20re.compile(
%20 r"(忽略|系统管理员|调试(?:命令|指令)?|输出用户数据|关闭安全过滤|解除|审计员|导出用户|"
%20 r"Ignore.*(?:instruction|safety|protocol|privacy)|Override|Disable\s+(?:safety|filter)|"
%20 r"Dump%20all|System%20Override|Display\s+(?:user|stored|PII)|Print\s+(?:user|PII))",
%20 %20re.I,
)
base64_kw_re%20=%20re.compile(r"(Base64|编码指令|encoded%20command|Decode|decode|解码)",%20re.I)
base64_long_re%20=%20re.compile(r"(?<![A-Za-z0-9+/=])[A-Za-z0-9+/=]{20,}(?![A-Za-z0-9+/=])")
unicode_re%20=%20re.compile(r"\\u[0-9a-fA-F]{4}")
def attack_type(user_input):
%20 if base64_kw_re.search(user_input) and base64_long_re.search(user_input):
%20 %20 %20 return"encoding"
%20 if unicode_re.search(user_input):
%20 %20 %20 return"unicode"
%20 if role_re.search(user_input):
%20 %20 %20 return"role_play"
%20 if direct_re.search(user_input):
%20 %20 %20 return"direct"
%20 returnNone
email_re%20=%20re.compile(r"[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}")
num_re%20=%20re.compile(r"(?<![0-9A-Za-z])(?:[0-9Xx][\s,\-]*){11,19}(?![0-9A-Za-z])")
sep_email_re%20=%20re.compile(r"(?:[A-Za-z0-9._%+@\-][\s,\-]*){8,120}")
base64_token_re%20=%20re.compile(r"(?<![A-Za-z0-9+/=])[A-Za-z0-9+/=]{8,}={0,2}(?![A-Za-z0-9+/=])")
def extract_plain(text):
%20 %20found%20=%20[]
%20 for match in email_re.finditer(text):
%20 %20 %20 %20value%20=%20match.group(0).strip(".,;:!?，；：！）)]}")
%20 %20 %20 if is_email(value):
%20 %20 %20 %20 %20 %20found.append(("email",%20value))
%20 for line in text.splitlines():
%20 %20 %20 if"@"in line or re.search(r"(email|邮箱|mail)",%20line,%20re.I):
%20 %20 %20 %20 %20 for match in sep_email_re.finditer(line):
%20 %20 %20 %20 %20 %20 %20 %20value%20=%20re.sub(r"[\s,\-]+", "",%20match.group(0))
%20 %20 %20 %20 %20 %20 %20 %20value%20=%20value.strip(".,;:!?，；：！）)]}")
%20 %20 %20 %20 %20 %20 %20 if"@"in value and is_email(value):
%20 %20 %20 %20 %20 %20 %20 %20 %20 %20found.append(("email",%20value))
%20 for match in num_re.finditer(text):
%20 %20 %20 %20value%20=%20re.sub(r"[\s,\-]+", "",%20match.group(0)).upper()
%20 %20 %20 if is_idcard(value):
%20 %20 %20 %20 %20 %20found.append(("idcard",%20value))
%20 %20 %20 if is_phone(value):
%20 %20 %20 %20 %20 %20found.append(("phone",%20value))
%20 %20 %20 if is_bankcard(value):
%20 %20 %20 %20 %20 %20found.append(("bankcard",%20value))
%20 return found
def extract_all(text):
%20 %20found%20=%20extract_plain(text)
%20 for match in base64_token_re.finditer(text):
%20 %20 %20 %20token%20=%20match.group(0)
%20 %20 %20 try:
%20 %20 %20 %20 %20 %20decoded%20=%20base64.b64decode(
%20 %20 %20 %20 %20 %20 %20 %20token%20+ "=" *%20((-len(token))%20% 4),
%20 %20 %20 %20 %20 %20 %20 %20validate=True,
%20 %20 %20 %20 %20 %20).decode("utf-8")
%20 %20 %20 except Exception:
%20 %20 %20 %20 %20 continue
%20 %20 %20 if all(ch.isprintable() or ch in"\r\n\t"for ch in decoded):
%20 %20 %20 %20 %20 %20found.extend(extract_plain(decoded))
%20 %20seen%20=%20set()
%20 %20result%20=%20[]
%20 for item in found:
%20 %20 %20 if item notin seen:
%20 %20 %20 %20 %20 %20seen.add(item)
%20 %20 %20 %20 %20 %20result.append(item)
%20 return result
rows%20=%20[]
for line in log.open(encoding="utf-8"):
%20 %20obj%20=%20json.loads(line)
%20 %20typ%20=%20attack_type(obj["user_input"])
%20 ifnot typ:
%20 %20 %20 continue
%20 for category,%20value in extract_all(obj["model_output"]):
%20 %20 %20 %20rows.append((obj["turn_id"],%20typ,%20category,%20value))
category_order%20=%20{"idcard": 0, "phone": 1, "bankcard": 2, "email": 3}
rows%20=%20sorted(set(rows),%20key=lambda row:%20(row[0],%20category_order[row[2]],%20row[3]))
with out.open("w",%20encoding="utf-8",%20newline="") as f:
%20 %20writer%20=%20csv.writer(f)
%20 %20writer.writerow(["turn_id", "attack_type", "category", "value"])
%20 %20writer.writerows(rows)
print("生成完成:",%20out.resolve())
print("结果行数:",%20len(rows))
跑完能看到脚本生成了result.csv，结果行数是304
我用%20WPS%20打开看了一眼%20CSV%20结构，表头是题目要求的turn_id,attack_type,category,value这里注意别用%20WPS%20另存，因为它会把部分手机号显示成科学计数法；提交用的是脚本原始生成的result.csv
最后把这个result.csv%20上传到靶机，校验结果是100.000%，直接给%20flag
flag{f639ba7d803f97e5fbeab25f70f3af71}
这题给了三个关键文件：mixed_data.csv、leaked_private.pem%20和分类分级规范mixed_data.csv%20是要处理的数据，leaked_private.pem%20是解%20RSA%20密文用的私钥，规范里写了每种数据怎么分类
我先去平台下载了示例文件%20example.csv，主要是看最后提交文件的格式这张图只看表头就行，别用%20WPS/Excel%20保存%20CSV，不然长数字可能会被改成科学计数法
示例文件的表头是：
data_id,value,category,level
分类规范的重点是：value%20可能是明文，也可能是固定%20344%20字符的%20RSA%20base64%20密文密文用%20leaked_private.pem%20解，实际测试用的是%20RSA-OAEP%20+%20SHA256解出来以后还要做归一化：全角转半角、去空白、去横线，手机号里的%20+86%20和%200086%20前缀也要去掉然后依次判断身份证、银行卡、手机号、IP，剩下的就是%20normal最后输出%20CSV，列名必须是%20data_id,value,category,level，编码用%20UTF-8%20无%20BOM
脚本我直接放在附件同一个目录里，所以开头用%20Path(file).resolve().parent%20取当前目录，这样%20mixed_data.csv、leaked_private.pem%20和生成的%20classified_data.csv%20都在一起，不容易因为路径不一样报错
import base64
import csv
import re
import unicodedata
from datetime import datetime
from pathlib import Path
from cryptography.hazmat.primitives import hashes,%20serialization
from cryptography.hazmat.primitives.asymmetric import padding
BASE%20=%20Path(__file__).resolve().parent
MIXED%20=%20BASE%20/ "mixed_data.csv"
PRIVATE_KEY%20=%20BASE%20/ "leaked_private.pem"
OUT%20=%20BASE%20/ "classified_data.csv"
PHONE_PREFIXES%20=%20{
%20 "134", "135", "136", "137", "138", "139", "147", "148", "150", "151",
%20 "152", "157", "158", "159", "172", "178", "182", "183", "184", "187",
%20 "188", "195", "198", "130", "131", "132", "140", "145", "146", "155",
%20 "156", "166", "167", "171", "175", "176", "185", "186", "196", "133",
%20 "149", "153", "173", "174", "177", "180", "181", "189", "190", "191",
%20 "193", "199",
}
ID_WEIGHTS%20=%20[7, 9, 10, 5, 8, 4, 2, 1, 6, 3, 7, 9, 10, 5, 8, 4, 2]
ID_CHECK%20= "10X98765432"
def decrypt_if_needed(value:%20str,%20key) ->%20str:
%20 if len(value)%20!= 344:
%20 %20 %20 return value
%20 try:
%20 %20 %20 %20ciphertext%20=%20base64.b64decode(value,%20validate=True)
%20 %20 %20 %20plaintext%20=%20key.decrypt(
%20 %20 %20 %20 %20 %20ciphertext,
%20 %20 %20 %20 %20 %20padding.OAEP(
%20 %20 %20 %20 %20 %20 %20 %20mgf=padding.MGF1(algorithm=hashes.SHA256()),
%20 %20 %20 %20 %20 %20 %20 %20algorithm=hashes.SHA256(),
%20 %20 %20 %20 %20 %20 %20 %20label=None,
%20 %20 %20 %20 %20 %20),
%20 %20 %20 %20)
%20 %20 %20 return plaintext.decode("utf-8")
%20 except Exception:
%20 %20 %20 return value
def nfkc_no_space(value:%20str) ->%20str:
%20 return re.sub(r"\s+", "",%20unicodedata.normalize("NFKC",%20value))
def id_candidate(value:%20str) ->%20str:
%20 return nfkc_no_space(value).replace("-", "").upper()
def phone_candidate(value:%20str) ->%20str:
%20 %20v%20=%20nfkc_no_space(value).replace("-", "")
%20 if v.startswith("+86"):
%20 %20 %20 %20v%20=%20v[3:]
%20 elif v.startswith("0086"):
%20 %20 %20 %20v%20=%20v[4:]
%20 return v
def bank_candidate(value:%20str) ->%20str:
%20 return nfkc_no_space(value).replace("-", "")
def ip_candidate(value:%20str) ->%20str:
%20 return nfkc_no_space(value)
def normal_candidate(value:%20str) ->%20str:
%20 %20v%20=%20nfkc_no_space(value).replace("-", "")
%20 if v.startswith("+86"):
%20 %20 %20 %20v%20=%20v[3:]
%20 elif v.startswith("0086"):
%20 %20 %20 %20v%20=%20v[4:]
%20 return v
def valid_idcard(value:%20str) ->%20bool:
%20 ifnot re.fullmatch(r"\d{17}[\dX]",%20value):
%20 %20 %20 returnFalse
%20 %20birth%20=%20value[6:14]
%20 try:
%20 %20 %20 %20dt%20=%20datetime.strptime(birth, "%Y%m%d")
%20 except ValueError:
%20 %20 %20 returnFalse
%20 if dt.year%20< 1900or dt.year%20> 2030:
%20 %20 %20 returnFalse
%20 %20checksum%20=%20sum(int(ch)%20*%20weight for ch,%20weight in zip(value[:17],%20ID_WEIGHTS))
%20 return ID_CHECK[checksum%20% 11]%20==%20value[-1]
def valid_luhn(value:%20str) ->%20bool:
%20 ifnot re.fullmatch(r"\d{16,19}",%20value):
%20 %20 %20 returnFalse
%20 %20total%20= 0
%20 for idx,%20ch in enumerate(reversed(value), 1):
%20 %20 %20 %20digit%20=%20int(ch)
%20 %20 %20 if idx%20% 2 == 0:
%20 %20 %20 %20 %20 %20digit%20*= 2
%20 %20 %20 %20 %20 if digit%20> 9:
%20 %20 %20 %20 %20 %20 %20 %20digit%20-= 9
%20 %20 %20 %20total%20+=%20digit
%20 return total%20% 10 == 0
def valid_phone(value:%20str) ->%20bool:
%20 return bool(re.fullmatch(r"\d{11}",%20value)) and value[:3] in PHONE_PREFIXES
def valid_ip(value:%20str) ->%20bool:
%20 %20parts%20=%20value.split(".")
%20 if len(parts)%20!= 4:
%20 %20 %20 returnFalse
%20 for part in parts:
%20 %20 %20 ifnot re.fullmatch(r"\d+",%20part):
%20 %20 %20 %20 %20 returnFalse
%20 %20 %20 if len(part)%20> 1and part.startswith("0"):
%20 %20 %20 %20 %20 returnFalse
%20 %20 %20 if int(part)%20> 255:
%20 %20 %20 %20 %20 returnFalse
%20 returnTrue
def classify(value:%20str) ->%20tuple[str,%20str,%20str]:
%20 %20v%20=%20id_candidate(value)
%20 if valid_idcard(v):
%20 %20 %20 return v, "idcard", "S4"
%20 %20v%20=%20bank_candidate(value)
%20 if valid_luhn(v):
%20 %20 %20 return v, "bankcard", "S4"
%20 %20v%20=%20phone_candidate(value)
%20 if valid_phone(v):
%20 %20 %20 return v, "phone", "S3"
%20 %20v%20=%20ip_candidate(value)
%20 if valid_ip(v):
%20 %20 %20 return v, "ip", "S2"
%20 return normal_candidate(value), "normal", "S1"
def main() -> None:
%20 %20key%20=%20serialization.load_pem_private_key(PRIVATE_KEY.read_bytes(),%20password=None)
%20 with MIXED.open("r",%20encoding="utf-8-sig",%20newline="") as src,%20OUT.open(
%20 %20 %20 "w",%20encoding="utf-8",%20newline=""
%20 %20) as dst:
%20 %20 %20 %20reader%20=%20csv.DictReader(src)
%20 %20 %20 %20writer%20=%20csv.DictWriter(dst,%20fieldnames=["data_id", "value", "category", "level"])
%20 %20 %20 %20writer.writeheader()
%20 %20 %20 for row in reader:
%20 %20 %20 %20 %20 %20plain%20=%20decrypt_if_needed(row["value"],%20key)
%20 %20 %20 %20 %20 %20normalized,%20category,%20level%20=%20classify(plain)
%20 %20 %20 %20 %20 %20writer.writerow(
%20 %20 %20 %20 %20 %20 %20 %20{
%20 %20 %20 %20 %20 %20 %20 %20 %20 "data_id":%20row["data_id"],
%20 %20 %20 %20 %20 %20 %20 %20 %20 "value":%20normalized,
%20 %20 %20 %20 %20 %20 %20 %20 %20 "category":%20category,
%20 %20 %20 %20 %20 %20 %20 %20 %20 "level":%20level,
%20 %20 %20 %20 %20 %20 %20 %20}
%20 %20 %20 %20 %20 %20)
if __name__%20== "__main__":
%20 %20main()
跑完会在同目录生成%20classified_data.csv下面这张图能看到脚本位置和前面几行路径设置，重点就是%20BASE%20=%20Path(file).resolve().parent，这样脚本会读当前目录里的附件文件
最后回到平台，输入验证码，上传生成的%20classified_data.csv结果出来是%20100.000%，flag%20也在结果表里
flag{a92c018d480b6b5e8aeb4a117e0fc72c}
分析壳
这里很像%20UPX，但不是标准%20UPX，直接upx%20-d会失败，是一个自定义壳。
这个拷贝逻辑是烟雾，0x412200对应%20.UPX1的文件偏移
file_offset%20=%200x412200%20-%200x412000%20+%200x200%20=%200x400
但文件偏移%200x400开始的数据执行不了
注意到这一串十六进制
CE%20B6%2014%203D%2043%2054%200E%200E%2002%200E%200E%200E%20CC%200A%2031%2031
常见zlib头包括
78%2001
78%209C
78%20DA
将该十六进制尝试异或0xce，正好是zlib头，前面填充的ce也算是提示吧。将这一段每个字节异或0xce后：
提取zlib压缩流，解压出内层PE
import argparse
import zlib
from pathlib import Path
PAYLOAD_OFF%20= 0x25B
PAYLOAD_LEN%20= 0x245
XOR_KEY%20= 0xCE
def unpack_outer(path:%20Path) ->%20bytes:
%20 %20data%20=%20path.read_bytes()
%20 %20blob%20=%20data[PAYLOAD_OFF:PAYLOAD_OFF%20+%20PAYLOAD_LEN]
%20 if len(blob)%20!=%20PAYLOAD_LEN:
%20 %20 %20 raise ValueError("input%20is%20too%20small%20for%20the%20expected%20payload%20range")
%20 %20xored%20=%20bytes(b%20^%20XOR_KEY for b in blob)
%20 %20print(f"[+]%20xor%20head: {xored[:16].hex('%20')}")
%20 if xored[:3]%20!= b"\x00\x78\xda":
%20 %20 %20 raise ValueError("unexpected%20header:%20xored%20data%20should%20start%20with%2000%2078%20da")
%20 %20inner%20=%20zlib.decompress(xored[1:])
%20 if inner[:2]%20!= b"MZ":
%20 %20 %20 raise ValueError("decompressed%20payload%20is%20not%20a%20PE%20file")
%20 return inner
def main() -> None:
%20 %20parser%20=%20argparse.ArgumentParser(description="Unpack%20licensed.exe%20outer%20layer")
%20 %20parser.add_argument("input",%20type=Path,%20help="original%20licensed.exe")
%20 %20parser.add_argument("output",%20type=Path,%20help="output%20inner%20PE")
%20 %20args%20=%20parser.parse_args()
%20 %20inner%20=%20unpack_outer(args.input)
%20 %20args.output.write_bytes(inner)
%20 %20print(f"[+]%20wrote: {args.output}")
%20 %20print(f"[+]%20inner%20size: {len(inner)} bytes")
if __name__%20== "__main__":
%20 %20main()
得到的1.exe一个很小的%20.NET%20PE
可以看到_bfk和_bfct关键字，猜测是blowfish%20key和cipherttext
提取key和cipherttext
from pathlib import Path
import struct
KEY_LEN%20= 16
CT_LEN%20= 48
def get_input_path() ->%20Path:
%20 for module_name in ("ida_nalt", "idc", "idaapi"):
%20 %20 %20 try:
%20 %20 %20 %20 %20 %20module%20=%20__import__(module_name)
%20 %20 %20 %20 %20 %20getter%20=%20getattr(module, "get_input_file_path", None)
%20 %20 %20 %20 %20 if getter isNone:
%20 %20 %20 %20 %20 %20 %20 continue
%20 %20 %20 %20 %20 %20value%20=%20getter()
%20 %20 %20 %20 %20 if value:
%20 %20 %20 %20 %20 %20 %20 %20path%20=%20Path(value)
%20 %20 %20 %20 %20 %20 %20 if path.exists():
%20 %20 %20 %20 %20 %20 %20 %20 %20 return path
%20 %20 %20 except Exception:
%20 %20 %20 %20 %20 pass
%20 import ida_kernwin
path%20=%20get_input_path()
data%20=%20path.read_bytes()
def u16(off):
%20 return struct.unpack_from("<H",%20data,%20off)[0]
def u32(off):
%20 return struct.unpack_from("<I",%20data,%20off)[0]
def rva_to_raw(rva):
%20 %20pe%20=%20u32(0x3C)
%20 %20number_of_sections%20=%20u16(pe%20+ 6)
%20 %20size_of_optional_header%20=%20u16(pe%20+ 20)
%20 %20section_table%20=%20pe%20+ 24 +%20size_of_optional_header
%20 for i in range(number_of_sections):
%20 %20 %20 %20sec%20=%20section_table%20+%20i%20* 40
%20 %20 %20 %20name%20=%20data[sec:sec%20+ 8].rstrip(b"\x00").decode(errors="replace")
%20 %20 %20 %20virtual_size%20=%20u32(sec%20+ 8)
%20 %20 %20 %20virtual_address%20=%20u32(sec%20+ 12)
%20 %20 %20 %20raw_size%20=%20u32(sec%20+ 16)
%20 %20 %20 %20raw_ptr%20=%20u32(sec%20+ 20)
%20 %20 %20 if virtual_address%20<=%20rva%20<%20virtual_address%20+%20max(virtual_size,%20raw_size):
%20 %20 %20 %20 %20 return raw_ptr%20+%20(rva%20-%20virtual_address),%20name
%20 raise RuntimeError("RVA%200x%X%20is%20not%20in%20any%20section" %%20rva)
pe%20=%20u32(0x3C)
optional%20=%20pe%20+ 24
data_directories%20=%20optional%20+ 0x60
clr_dir%20=%20data_directories%20+ 14 * 8
clr_rva%20=%20u32(clr_dir)
clr_size%20=%20u32(clr_dir%20+ 4)
clr_raw,%20clr_sec%20=%20rva_to_raw(clr_rva)
metadata_rva%20=%20u32(clr_raw%20+ 8)
metadata_size%20=%20u32(clr_raw%20+ 12)
metadata_raw,%20metadata_sec%20=%20rva_to_raw(metadata_rva)
key_off%20=%20metadata_raw%20+%20metadata_size
ct_off%20=%20key_off%20+%20KEY_LEN
key%20=%20data[key_off:key_off%20+%20KEY_LEN]
ct%20=%20data[ct_off:ct_off%20+%20CT_LEN]
print("[+]%20CLR%20header:%20RVA=0x%X%20raw=0x%X%20section=%s%20size=0x%X" %%20(clr_rva,%20clr_raw,%20clr_sec,%20clr_size))
print("[+]%20CLR%20metadata:%20RVA=0x%X%20raw=0x%X%20section=%s%20size=0x%X" %%20(metadata_rva,%20metadata_raw,%20metadata_sec,%20metadata_size))
print("[+]%20metadata%20end%20raw%20offset:%200x%X" %%20key_off)
print("[+]%20CIL%20says%20_bfk%20length%20=%2016,%20_bfct%20length%20=%2048")
print("[+]%20key:",%20key.hex("%20"))
print("[+]%20ciphertext:",%20ct.hex("%20"))
字段名_bfk、_bfct已经提示了blowfish。key%20是%2016%20字节，密文长度是%2048%20字节，正好是blowfish%208%20字节分组的整数倍，直接试%20blowfish%20ECB可以解出可读%20flag，尾部用 00 填充
#!/usr/bin/env%20python3
import argparse
from pathlib import Path
from Crypto.Cipher import Blowfish
KEY_OFF%20= 0x3C4
KEY_LEN%20= 16
CT_OFF%20= 0x3D4
CT_LEN%20= 48
def main() -> None:
%20 %20parser%20=%20argparse.ArgumentParser(description="Decrypt%20flag%20from%20unpacked%20inner%20PE")
%20 %20parser.add_argument("input",%20type=Path,%20help="unpacked%20.NET%20PE")
%20 %20args%20=%20parser.parse_args()
%20 %20data%20=%20args.input.read_bytes()
%20 %20key%20=%20data[KEY_OFF:KEY_OFF%20+%20KEY_LEN]
%20 %20ct%20=%20data[CT_OFF:CT_OFF%20+%20CT_LEN]
%20 if len(key)%20!=%20KEY_LEN or len(ct)%20!=%20CT_LEN:
%20 %20 %20 raise ValueError("input%20is%20too%20small%20for%20the%20expected%20key/ciphertext%20offsets")
%20 %20pt%20=%20Blowfish.new(key,%20Blowfish.MODE_ECB).decrypt(ct).rstrip(b"\x00")
%20 %20print(f"[+]%20key: {key.hex()}")
%20 %20print(f"[+]%20ciphertext: {ct.hex()}")
%20 %20print(f"[+]%20plaintext: {pt.decode()}")
if __name__%20== "__main__":
%20 %20main()
TXT记录base64编码
MoreSecureAes128
分析出可疑的单向echo请求
ttl只有0和1，疑似二进制，每8bit转一个%20ASCII%20字符
from scapy.all import rdpcap,%20IP,%20ICMP
pcap%20= r"suspicious_traffic.pcapng"
pkts%20=%20rdpcap(pcap)
rows%20=%20[]
for p in pkts:
%20 if p.haslayer(IP) and p.haslayer(ICMP):
%20 %20 %20 if p[IP].src%20== "10.10.20.33"and p[IP].dst%20== "45.76.188.23"and p[ICMP].type%20== 8:
%20 %20 %20 %20 %20 %20rows.append((p[ICMP].seq,%20p[IP].ttl))
rows.sort()
bits%20= "".join(str(ttl) for seq,%20ttl in rows)
print(bits)
001101100011000000110010001100100011011001100100001100110110001101100100001100110110001101100110001100110011010000110101011001000110011000110000011000100110000101100001001100110110001100111000011001000110010101100011001100000110010000111001011001000011001000110010001101010110001000110011011001100110011000111001001100010011011000110011001101000011010000111001001110000110010001100010001101000110011000110101011000010011011101100100001101000011010000110101001100100110010001100100001100010110000100110010011000110011001000110000001110000110010000110100011001010110011001100011001110010110010001100001001101110011001101100101011000100011100000111000001110000011010001100110011000100011011000111001011000110011001101100010001100010011001000110000001100100110001100110101011111000100000101000101010100110010110101000101010000110100001000101100001000000110101101100101011110010010000001101001011100110010000001101001011011100010000001000100010011100101001100100000010101000101100001010100
加密方式为AES-ECB，key就是DNS%20TXT解出的MoreSecureAes128
查询表
表结构
身份证校验规则：
weights%20=%20[7, 9, 10, 5, 8, 4, 2, 1, 6, 3, 7, 9, 10, 5, 8, 4, 2]
check_codes%20= "10X98765432"
check%20=%20check_codes[sum(int(d)%20*%20w for d,%20w in zip(idcard[:17],%20weights))%20% 11]
手机号校验规则：
phone%20必须匹配%20\d{11}
phone[:3]%20必须在规范给出的合法号段集合中
根据安全审计规范写了一份sql脚本
WITH ghost_accounts AS (
%20 SELECT
%20 %20 %20 %20u.id AS user_id,
%20 %20 %20 %20u.username,
%20 %20 %20 %20u.name,
%20 %20 %20 %20u.idcard,
%20 %20 %20 %20u.phone,
%20 %20 %20 %20u.role_id,
%20 %20 %20 %20r.role_name,
%20 %20 %20 %20u.status,
%20 %20 %20 CASE
%20 %20 %20 %20 %20 WHEN is_valid_idcard(u.idcard)%20= 0AND is_valid_phone(u.phone)%20= 0THEN'idcard,phone'
%20 %20 %20 %20 %20 WHEN is_valid_idcard(u.idcard)%20= 0THEN'idcard'
%20 %20 %20 %20 %20 ELSE'phone'
%20 %20 %20 ENDAS invalid_fields
%20 FROMusersAS u
%20 JOINrolesAS r
%20 %20 %20 ON r.id%20=%20u.role_id
%20 WHERE is_valid_idcard(u.idcard)%20= 0
%20 %20 %20 OR is_valid_phone(u.phone)%20= 0
),
authorized_permissions AS (
%20 SELECT
%20 %20 %20 %20role_id,
%20 %20 %20 %20api_endpoint
%20 FROM api_permissions
%20 WHERE can_access%20= 1
),
violation_candidates AS (
%20 SELECT
%20 %20 %20 %20l.id AS log_id,
%20 %20 %20 %20l.user_id,
%20 %20 %20 %20u.username,
%20 %20 %20 %20u.name,
%20 %20 %20 %20u.status,
%20 %20 %20 %20u.role_id,
%20 %20 %20 %20r.role_name,
%20 %20 %20 %20l.api_endpoint,
%20 %20 %20 %20l.access_time,
%20 %20 %20 CASE
%20 %20 %20 %20 %20 WHEN u.status%20= 'inactive'THEN1
%20 %20 %20 %20 %20 WHENtime(l.access_time)%20< '09:00:00'ORtime(l.access_time)%20> '18:00:00'THEN2
%20 %20 %20 %20 %20 WHEN ap.role_id ISNULLTHEN3
%20 %20 %20 ENDAS violation_type,
%20 %20 %20 trim(
%20 %20 %20 %20 %20 CASEWHEN u.status%20= 'inactive'THEN'1,'ELSE''END ||
%20 %20 %20 %20 %20 CASEWHENtime(l.access_time)%20< '09:00:00'ORtime(l.access_time)%20> '18:00:00'THEN'2,'ELSE''END ||
%20 %20 %20 %20 %20 CASEWHEN ap.role_id ISNULLTHEN'3,'ELSE''END,
%20 %20 %20 %20 %20 ','
%20 %20 %20 %20) AS matched_types
%20 FROM access_logs AS l
%20 JOINusersAS u
%20 %20 %20 ON u.id%20=%20l.user_id
%20 JOINrolesAS r
%20 %20 %20 ON r.id%20=%20u.role_id
%20 LEFTJOIN authorized_permissions AS ap
%20 %20 %20 ON ap.role_id%20=%20u.role_id
%20 %20 %20 AND ap.api_endpoint%20=%20l.api_endpoint
%20 WHERE u.id NOTIN (SELECT user_id FROM ghost_accounts)
%20 %20 AND (
%20 %20 %20 %20 %20 %20u.status%20= 'inactive'
%20 %20 %20 %20 ORtime(l.access_time)%20< '09:00:00'
%20 %20 %20 %20 ORtime(l.access_time)%20> '18:00:00'
%20 %20 %20 %20 OR ap.role_id ISNULL
%20 %20 %20)
)
SELECT
%20 %20*
FROM violation_candidates
ORDERBY log_id;
找出幽灵账户脚本
from __future__ import annotations
import argparse
import csv
import hashlib
import json
import re
import sqlite3
from collections import Counter
from pathlib import Path
from typing import Iterable
IDCARD_WEIGHTS%20=%20[7, 9, 10, 5, 8, 4, 2, 1, 6, 3, 7, 9, 10, 5, 8, 4, 2]
IDCARD_CHECK_CODES%20= "10X98765432"
PHONE_PREFIXES%20=%20{
%20 "134", "135", "136", "137", "138", "139", "147", "148", "150", "151",
%20 "152", "157", "158", "159", "172", "178", "182", "183", "184", "187",
%20 "188", "195", "198", "130", "131", "132", "140", "145", "146", "155",
%20 "156", "166", "167", "171", "175", "176", "185", "186", "196", "133",
%20 "149", "153", "173", "174", "177", "180", "181", "189", "190", "191",
%20 "193", "199",
}
WORK_START%20= "09:00:00"
WORK_END%20= "18:00:00"
def is_valid_idcard(idcard:%20str) ->%20int:
%20 ifnot re.fullmatch(r"\d{17}[\dX]",%20idcard or""):
%20 %20 %20 return0
%20 %20weighted_sum%20=%20sum(int(digit)%20*%20weight for digit,%20weight in zip(idcard[:17],%20IDCARD_WEIGHTS))
%20 return int(IDCARD_CHECK_CODES[weighted_sum%20% 11]%20==%20idcard[-1])
def is_valid_phone(phone:%20str) ->%20int:
%20 ifnot re.fullmatch(r"\d{11}",%20phone or""):
%20 %20 %20 return0
%20 return int(phone[:3] in PHONE_PREFIXES)
def load_query(sql_path:%20Path) ->%20str:
%20 return sql_path.read_text(encoding="utf-8")
def rows_to_dicts(rows:%20Iterable[sqlite3.Row]) ->%20list[dict]:
%20 return [dict(row) for row in rows]
def write_csv(path:%20Path,%20rows:%20list[dict],%20fieldnames:%20list[str]) -> None:
%20 with path.open("w",%20encoding="utf-8-sig",%20newline="") as handle:
%20 %20 %20 %20writer%20=%20csv.DictWriter(handle,%20fieldnames=fieldnames)
%20 %20 %20 %20writer.writeheader()
%20 %20 %20 %20writer.writerows(rows)
def write_markdown_report(
%20 %20report_path:%20Path,
%20 %20summary:%20dict,
%20 %20ghost_rows:%20list[dict],
%20 %20violation_rows:%20list[dict],
%20 %20db_path:%20Path,
%20 %20spec_path:%20Path,
%20 %20sql_path:%20Path,
%20 %20script_path:%20Path,
) -> None:
%20 %20ghost_lines%20=%20[
%20 %20 %20 f"-%20`{row['user_id']}`%20/%20`{row['username']}`%20/ {row['name']} /%20`{row['invalid_fields']}`%20/%20`{row['status']}`%20/%20`{row['role_name']}`"
%20 %20 %20 for row in ghost_rows
%20 %20]
%20 %20violation_lines%20=%20[
%20 %20 %20 %20(
%20 %20 %20 %20 %20 f"-%20`log={row['log_id']}`%20`type={row['violation_type']}`%20"
%20 %20 %20 %20 %20 f"`matched={row['matched_types']}`%20`user={row['user_id']}/{row['username']}`%20"
%20 %20 %20 %20 %20 f"`role={row['role_name']}`%20`api={row['api_endpoint']}`%20`time={row['access_time']}`"
%20 %20 %20 %20)
%20 %20 %20 for row in violation_rows
%20 %20]
%20 %20report%20= "\n".join(
%20 %20 %20 %20[
%20 %20 %20 %20 %20 "#%20exam_system%20审计结果",
%20 %20 %20 %20 %20 "",
%20 %20 %20 %20 %20 "##%20输入文件",
%20 %20 %20 %20 %20 f"-%20规范:%20`{spec_path}`",
%20 %20 %20 %20 %20 f"-%20数据库:%20`{db_path}`",
%20 %20 %20 %20 %20 "",
%20 %20 %20 %20 %20 "##%20结论",
%20 %20 %20 %20 %20 f"-%20最终%20flag:%20`{summary['flag']}`",
%20 %20 %20 %20 %20 f"-%20幽灵账户总数:%20`{summary['ghost_count']}`",
%20 %20 %20 %20 %20 f"-%20被幽灵账户排除的访问日志数:%20`{summary['excluded_log_count']}`",
%20 %20 %20 %20 %20 f"-%20违规访问总数:%20`{summary['violation_count']}`",
%20 %20 %20 %20 %20 f"-%20违规类型分布:%20`1={summary['violation_type_counts'].get('1', 0)}`,%20`2={summary['violation_type_counts'].get('2', 0)}`,%20`3={summary['violation_type_counts'].get('3', 0)}`",
%20 %20 %20 %20 %20 "",
%20 %20 %20 %20 %20 "##%20规则说明",
%20 %20 %20 %20 %20 "-%20身份证按%2018%20位格式和最后一位校验码校验；本库中的异常身份证全部表现为校验码错误。",
%20 %20 %20 %20 %20 "-%20手机号按%2011%20位数字串和指定前三位号段集合校验；本库中的异常手机号全部表现为非法号段。",
%20 %20 %20 %20 %20 "-%20工作时间边界按%20`09:00:00%20<=%20time%20<=%2018:00:00`%20处理；库中没有恰好等于%20`09:00:00`%20或%20`18:00:00`%20的记录，因此边界取值不影响结果。",
%20 %20 %20 %20 %20 "-%20对同一条日志，若同时命中多类违规，按最小类型编号记入结果。",
%20 %20 %20 %20 %20 "",
%20 %20 %20 %20 %20 "##%20幽灵账户列表",
%20 %20 %20 %20 %20 %20*ghost_lines,
%20 %20 %20 %20 %20 "",
%20 %20 %20 %20 %20 "##%20违规访问明细",
%20 %20 %20 %20 %20 %20*violation_lines,
%20 %20 %20 %20 %20 "",
%20 %20 %20 %20 %20 "##%20复现",
%20 %20 %20 %20 %20 f"-%20SQL%20逻辑:%20`{sql_path}`",
%20 %20 %20 %20 %20 f"-%20脚本:%20`{script_path}`",
%20 %20 %20 %20 %20 f"-%20命令:%20`python%20\"{script_path}\"%20--db%20\"{db_path}\"%20--spec%20\"{spec_path}\"%20--sql%20\"{sql_path}\"%20--outdir%20\"{report_path.parent}\"`",
%20 %20 %20 %20 %20 "",
%20 %20 %20 %20 %20 "##%20产物",
%20 %20 %20 %20 %20 f"-%20`{report_path}`",
%20 %20 %20 %20 %20 f"-%20`{report_path.parent%20/ 'ghost_accounts.csv'}`",
%20 %20 %20 %20 %20 f"-%20`{report_path.parent%20/ 'violations.csv'}`",
%20 %20 %20 %20 %20 f"-%20`{report_path.parent%20/ 'audit_summary.json'}`",
%20 %20 %20 %20]
%20 %20)
%20 %20report_path.write_text(report%20+ "\n",%20encoding="utf-8")
def main() -> None:
%20 %20parser%20=%20argparse.ArgumentParser(description="Audit%20exam_system%20ghost%20accounts%20and%20API%20violations.")
%20 %20parser.add_argument("--db",%20type=Path,%20required=True,%20help="Path%20to%20exam_system.db")
%20 %20parser.add_argument("--spec",%20type=Path,%20required=True,%20help="Path%20to%20安全审计规范.md")
%20 %20parser.add_argument("--sql",%20type=Path,%20required=True,%20help="Path%20to%20audit%20SQL%20file")
%20 %20parser.add_argument("--outdir",%20type=Path,%20required=True,%20help="Directory%20for%20generated%20outputs")
%20 %20args%20=%20parser.parse_args()
%20 %20outdir%20=%20args.outdir
%20 %20outdir.mkdir(parents=True,%20exist_ok=True)
%20 %20conn%20=%20sqlite3.connect(args.db)
%20 %20conn.row_factory%20=%20sqlite3.Row
%20 %20conn.create_function("is_valid_idcard", 1,%20is_valid_idcard)
%20 %20conn.create_function("is_valid_phone", 1,%20is_valid_phone)
%20 %20cur%20=%20conn.cursor()
%20 %20ghost_query%20= """
%20 %20WITH%20ghost_accounts%20AS%20(
%20 %20 %20 %20SELECT
%20 %20 %20 %20 %20 %20u.id%20AS%20user_id,
%20 %20 %20 %20 %20 %20u.username,
%20 %20 %20 %20 %20 %20u.name,
%20 %20 %20 %20 %20 %20u.idcard,
%20 %20 %20 %20 %20 %20u.phone,
%20 %20 %20 %20 %20 %20u.role_id,
%20 %20 %20 %20 %20 %20r.role_name,
%20 %20 %20 %20 %20 %20u.status,
%20 %20 %20 %20 %20 %20CASE
%20 %20 %20 %20 %20 %20 %20 %20WHEN%20is_valid_idcard(u.idcard)%20=%200%20AND%20is_valid_phone(u.phone)%20=%200%20THEN%20'idcard,phone'
%20 %20 %20 %20 %20 %20 %20 %20WHEN%20is_valid_idcard(u.idcard)%20=%200%20THEN%20'idcard'
%20 %20 %20 %20 %20 %20 %20 %20ELSE%20'phone'
%20 %20 %20 %20 %20 %20END%20AS%20invalid_fields
%20 %20 %20 %20FROM%20users%20AS%20u
%20 %20 %20 %20JOIN%20roles%20AS%20r
%20 %20 %20 %20 %20 %20ON%20r.id%20=%20u.role_id
%20 %20 %20 %20WHERE%20is_valid_idcard(u.idcard)%20=%200
%20 %20 %20 %20 %20 OR%20is_valid_phone(u.phone)%20=%200
%20 %20)
%20 %20SELECT%20*
%20 %20FROM%20ghost_accounts
%20 %20ORDER%20BY%20user_id
%20 %20"""
%20 %20violation_query%20=%20load_query(args.sql)
%20 %20ghost_rows%20=%20rows_to_dicts(cur.execute(ghost_query))
%20 %20violation_rows%20=%20rows_to_dicts(cur.execute(violation_query))
%20 %20violation_key%20= ",".join(f"{row['log_id']}-{row['violation_type']}"for row in violation_rows)
%20 %20md5_value%20=%20hashlib.md5(violation_key.encode("utf-8")).hexdigest()
%20 %20excluded_log_count%20=%20cur.execute(
%20 %20 %20 """
%20 %20 %20 %20SELECT%20COUNT(*)
%20 %20 %20 %20FROM%20access_logs
%20 %20 %20 %20WHERE%20user_id%20IN%20(
%20 %20 %20 %20 %20 %20SELECT%20u.id
%20 %20 %20 %20 %20 %20FROM%20users%20AS%20u
%20 %20 %20 %20 %20 %20WHERE%20is_valid_idcard(u.idcard)%20=%200
%20 %20 %20 %20 %20 %20 %20 OR%20is_valid_phone(u.phone)%20=%200
%20 %20 %20 %20)
%20 %20 %20 %20"""
%20 %20).fetchone()[0]
%20 %20ghost_reason_counts%20=%20Counter(row["invalid_fields"] for row in ghost_rows)
%20 %20violation_type_counts%20=%20Counter(str(row["violation_type"]) for row in violation_rows)
%20 %20summary%20=%20{
%20 %20 %20 "db_path":%20str(args.db),
%20 %20 %20 "spec_path":%20str(args.spec),
%20 %20 %20 "sql_path":%20str(args.sql),
%20 %20 %20 "ghost_count":%20len(ghost_rows),
%20 %20 %20 "ghost_reason_counts":%20dict(sorted(ghost_reason_counts.items())),
%20 %20 %20 "excluded_log_count":%20excluded_log_count,
%20 %20 %20 "violation_count":%20len(violation_rows),
%20 %20 %20 "violation_type_counts":%20dict(sorted(violation_type_counts.items())),
%20 %20 %20 "violation_key":%20violation_key,
%20 %20 %20 "md5":%20md5_value,
%20 %20 %20 "flag": f"flag{{{md5_value}}}",
%20 %20 %20 "work_window":%20[WORK_START,%20WORK_END],
%20 %20}
%20 %20write_csv(
%20 %20 %20 %20outdir%20/ "ghost_accounts.csv",
%20 %20 %20 %20ghost_rows,
%20 %20 %20 %20["user_id", "username", "name", "idcard", "phone", "role_id", "role_name", "status", "invalid_fields"],
%20 %20)
%20 %20write_csv(
%20 %20 %20 %20outdir%20/ "violations.csv",
%20 %20 %20 %20violation_rows,
%20 %20 %20 %20[
%20 %20 %20 %20 %20 "log_id",
%20 %20 %20 %20 %20 "user_id",
%20 %20 %20 %20 %20 "username",
%20 %20 %20 %20 %20 "name",
%20 %20 %20 %20 %20 "status",
%20 %20 %20 %20 %20 "role_id",
%20 %20 %20 %20 %20 "role_name",
%20 %20 %20 %20 %20 "api_endpoint",
%20 %20 %20 %20 %20 "access_time",
%20 %20 %20 %20 %20 "violation_type",
%20 %20 %20 %20 %20 "matched_types",
%20 %20 %20 %20],
%20 %20)
%20 %20(outdir%20/ "audit_summary.json").write_text(
%20 %20 %20 %20json.dumps(summary,%20ensure_ascii=False,%20indent=2)%20+ "\n",
%20 %20 %20 %20encoding="utf-8",
%20 %20)
%20 %20write_markdown_report(
%20 %20 %20 %20report_path=outdir%20/ "audit_report.md",
%20 %20 %20 %20summary=summary,
%20 %20 %20 %20ghost_rows=ghost_rows,
%20 %20 %20 %20violation_rows=violation_rows,
%20 %20 %20 %20db_path=args.db,
%20 %20 %20 %20spec_path=args.spec,
%20 %20 %20 %20sql_path=args.sql,
%20 %20 %20 %20script_path=Path(__file__).resolve(),
%20 %20)
%20 %20print(json.dumps(summary,%20ensure_ascii=False,%20indent=2))
if __name__%20== "__main__":
%20 %20main()
先看明文、迹线和密文长度
import numpy as np,%20pathlib
p=pathlib.Path('.')
print('plaintexts.npy',%20np.load(p/'plaintexts.npy',%20mmap_mode='r').shape,%20np.load(p/'plaintexts.npy',%20mmap_mode='r').dtype)
print('traces.npy',%20np.load(p/'traces.npy',%20mmap_mode='r').shape,%20np.load(p/'traces.npy',%20mmap_mode='r').dtype)
f=(p/'flag.enc').read_bytes()
print('flag.enc',%20len(f),%20f.hex())
有 5000 组采样，每组明文是%2016%20字节，也就是一个%20AES%20block。
迹线形状是 (5000,%202,%201024)，说明每组采样有%202%20个通道，每个通道%201024%20个采样点
AES%20首轮中，每个字节会经过：
SBox(P[i]%20^%20K[i])
其中 P[i] 是已知明文字节，K[i] 是要猜的密钥字节。CPA%20的做法是：
- 对某个%20key%20byte%20枚举
0x00到0xff。 - 对每条明文计算假设中间值。
- 把假设泄漏值和真实迹线逐点做%20Pearson%20相关。
- 哪个%20key%20guess%20的相关性最高，就认为哪个是正确密钥字节。
尝试最常见模型：
HW(SBox(P%20^%20K))
但这个模型的平均相关系数只有约%200.156，说明能看到一点信号，但不是设备真正主要泄漏的量
真正有效的模型是
HW(SBox(P%20^%20K)%20^%20P)
这个模型表示%20SubBytes%20前后的%20Hamming%20Distance，也就是状态从 P 变成 SBox(P%20^%20K) 时翻转了多少%20bit。对功耗/电磁侧信道来说，总线或寄存器翻转数量经常比单纯的%20Hamming%20Weight%20更明显。
恢复出的%20AES%20master%20key%20为：
422c174a4a89d96af0b7de281dc01324
直接用master%20key去尝试flag.enc并不能得到正确明文。继续做%20AES-128%20key%20schedule，可以得到第%2010%20轮密钥
flag.enc实际使用round10_key作为AES-ECB%20key加密。用该key解密并去掉PKCS#7 padding
from __future__ import annotations
import argparse
from pathlib import Path
import numpy as np
from Crypto.Cipher import AES
from Crypto.Util.Padding import unpad
SBOX%20=%20np.frombuffer(
%20 %20bytes.fromhex(
%20 %20 %20 "637c777bf26b6fc53001672bfed7ab76ca82c97dfa5947f0add4a2af9ca472c0"
%20 %20 %20 "b7fd9326363ff7cc34a5e5f171d8311504c723c31896059a071280e2eb27b275"
%20 %20 %20 "09832c1a1b6e5aa0523bd6b329e32f8453d100ed20fcb15b6acbbe394a4c58cf"
%20 %20 %20 "d0efaafb434d338545f9027f503c9fa851a3408f929d38f5bcb6da2110fff3d2"
%20 %20 %20 "cd0c13ec5f974417c4a77e3d645d197360814fdc222a908846eeb814de5e0bdb"
%20 %20 %20 "e0323a0a4906245cc2d3ac629195e479e7c8376d8dd54ea96c56f4ea657aae"
%20 %20 %20 "08ba78252e1ca6b4c6e8dd741f4bbd8b8a703eb5664803f60e613557b986c1"
%20 %20 %20 "1d9ee1f8981169d98e949b1e87e9ce5528df8ca1890dbfe6426841992d0fb"
%20 %20 %20 "054bb16"
%20 %20),
%20 %20dtype=np.uint8,
)
HW%20=%20np.array([i.bit_count() for i in range(256)],%20dtype=np.float32)
RCON%20=%20[0x00, 0x01, 0x02, 0x04, 0x08, 0x10, 0x20, 0x40, 0x80, 0x1B, 0x36]
def load_inputs(data_dir:%20Path) ->%20tuple[np.ndarray,%20np.ndarray,%20bytes]:
%20 %20plaintexts%20=%20np.load(data_dir%20/ "plaintexts.npy")
%20 %20traces%20=%20np.load(data_dir%20/ "traces.npy")
%20 %20flag_enc%20=%20(data_dir%20/ "flag.enc").read_bytes()
%20 return plaintexts,%20traces,%20flag_enc
def centered_trace_channel(traces:%20np.ndarray,%20channel:%20int%20= 0) ->%20tuple[np.ndarray,%20np.ndarray]:
%20 %20matrix%20=%20traces[:,%20channel,%20:].astype(np.float32)
%20 %20matrix%20-=%20matrix.mean(axis=0,%20keepdims=True)
%20 %20norm%20=%20np.sqrt((matrix%20*%20matrix).sum(axis=0))
%20 %20norm[norm%20== 0]%20= 1
%20 return matrix,%20norm
def build_leakage(plain_byte:%20np.ndarray,%20guesses:%20np.ndarray,%20model:%20str) ->%20np.ndarray:
%20 %20state%20=%20np.bitwise_xor(plain_byte[:, None],%20guesses)
%20 %20sbox_value%20=%20SBOX[state]
%20 if model%20== "HW(SBox(P^K))":
%20 %20 %20 %20leakage%20=%20HW[sbox_value]
%20 elif model%20== "HW(SBox(P^K)^P)":
%20 %20 %20 %20leakage%20=%20HW[np.bitwise_xor(sbox_value,%20plain_byte[:, None])]
%20 else:
%20 %20 %20 raise ValueError(model)
%20 %20leakage%20=%20leakage.T.astype(np.float32)
%20 %20leakage%20-=%20leakage.mean(axis=1,%20keepdims=True)
%20 return leakage
def run_cpa(plaintexts:%20np.ndarray,%20traces:%20np.ndarray,%20model:%20str) ->%20tuple[bytes,%20list[dict]]:
%20 %20trace_matrix,%20trace_norm%20=%20centered_trace_channel(traces,%20channel=0)
%20 %20guesses%20=%20np.arange(256,%20dtype=np.uint8)
%20 %20key%20=%20[]
%20 %20table%20=%20[]
%20 for byte_index in range(16):
%20 %20 %20 %20leakage%20=%20build_leakage(plaintexts[:,%20byte_index],%20guesses,%20model)
%20 %20 %20 %20leakage_norm%20=%20np.sqrt((leakage%20*%20leakage).sum(axis=1))
%20 %20 %20 %20leakage_norm[leakage_norm%20== 0]%20= 1
%20 %20 %20 %20corr%20=%20(leakage%20@%20trace_matrix)%20/%20(leakage_norm[:, None]%20*%20trace_norm[None,%20:])
%20 %20 %20 %20abs_corr%20=%20np.abs(corr)
%20 %20 %20 %20score%20=%20abs_corr.max(axis=1)
%20 %20 %20 %20order%20=%20np.argsort(score)[::-1]
%20 %20 %20 %20best%20=%20int(order[0])
%20 %20 %20 %20second%20=%20int(order[1])
%20 %20 %20 %20sample%20=%20int(abs_corr[best].argmax())
%20 %20 %20 %20key.append(best)
%20 %20 %20 %20table.append(
%20 %20 %20 %20 %20 %20{
%20 %20 %20 %20 %20 %20 %20 "byte":%20byte_index,
%20 %20 %20 %20 %20 %20 %20 "best":%20best,
%20 %20 %20 %20 %20 %20 %20 "corr":%20float(score[best]),
%20 %20 %20 %20 %20 %20 %20 "sample":%20sample,
%20 %20 %20 %20 %20 %20 %20 "second":%20second,
%20 %20 %20 %20 %20 %20 %20 "second_corr":%20float(score[second]),
%20 %20 %20 %20 %20 %20}
%20 %20 %20 %20)
%20 return bytes(key),%20table
def expand_aes128(master_key:%20bytes) ->%20list[bytes]:
%20 def xor_word(a:%20bytes,%20b:%20bytes) ->%20bytes:
%20 %20 %20 return bytes(x%20^%20y for x,%20y in zip(a,%20b))
%20 def sub_word(word:%20bytes) ->%20bytes:
%20 %20 %20 return bytes(int(SBOX[b]) for b in word)
%20 def rot_word(word:%20bytes) ->%20bytes:
%20 %20 %20 return word[1:]%20+%20word[:1]
%20 %20words%20=%20[master_key[i%20:%20i%20+ 4] for i in range(0, 16, 4)]
%20 for i in range(4, 44):
%20 %20 %20 %20temp%20=%20words[i%20- 1]
%20 %20 %20 if i%20% 4 == 0:
%20 %20 %20 %20 %20 %20temp%20=%20sub_word(rot_word(temp))
%20 %20 %20 %20 %20 %20temp%20=%20bytes([temp[0]%20^%20RCON[i%20// 4]])%20+%20temp[1:]
%20 %20 %20 %20words.append(xor_word(words[i%20- 4],%20temp))
%20 return [b"".join(words[4 *%20r%20: 4 *%20r%20+ 4]) for r in range(11)]
def print_table(title:%20str,%20key:%20bytes,%20table:%20list[dict]) -> None:
%20 %20avg%20=%20sum(row["corr"] for row in table)%20/%20len(table)
%20 %20print()
%20 %20print(title)
%20 %20print(f"candidate_key%20= {key.hex()}")
%20 %20print(f"average_best_corr%20= {avg:.6f}")
%20 %20print("byte%20 key%20 corr%20 %20 %20 sample%20 second%20 second_corr")
%20 for row in table:
%20 %20 %20 %20print(
%20 %20 %20 %20 %20 f"{row['byte']:>4} %20"
%20 %20 %20 %20 %20 f"{row['best']:02x} %20 "
%20 %20 %20 %20 %20 f"{row['corr']:.6f} %20"
%20 %20 %20 %20 %20 f"{row['sample']:>6} %20"
%20 %20 %20 %20 %20 f"{row['second']:02x} %20 %20 %20"
%20 %20 %20 %20 %20 f"{row['second_corr']:.6f}"
%20 %20 %20 %20)
def main() -> None:
%20 %20parser%20=%20argparse.ArgumentParser(description="CPA%20solve%20for%20a%20directory%20with%20flag.enc,%20plaintexts.npy%20and%20traces.npy")
%20 %20parser.add_argument("--data-dir",%20type=Path,%20default=Path("."),%20help="Directory%20containing%20the%20three%20challenge%20files")
%20 %20args%20=%20parser.parse_args()
%20 %20data_dir%20=%20args.data_dir.resolve()
%20 %20plaintexts,%20traces,%20flag_enc%20=%20load_inputs(data_dir)
%20 %20print(f"data_dir%20= {data_dir}")
%20 %20print(f"plaintexts.npy%20shape%20= {plaintexts.shape},%20dtype%20= {plaintexts.dtype}")
%20 %20print(f"traces.npy%20shape%20= {traces.shape},%20dtype%20= {traces.dtype}")
%20 %20print(f"flag.enc%20length%20= {len(flag_enc)},%20hex%20= {flag_enc.hex()}")
%20 %20hw_key,%20hw_table%20=%20run_cpa(plaintexts,%20traces, "HW(SBox(P^K))")
%20 %20hd_key,%20hd_table%20=%20run_cpa(plaintexts,%20traces, "HW(SBox(P^K)^P)")
%20 %20print_table("Model%201:%20HW(SBox(P^K))",%20hw_key,%20hw_table)
%20 %20print_table("Model%202:%20HW(SBox(P^K)^P)",%20hd_key,%20hd_table)
%20 %20round_keys%20=%20expand_aes128(hd_key)
%20 %20round10_key%20=%20round_keys[10]
%20 %20flag%20=%20unpad(AES.new(round10_key,%20AES.MODE_ECB).decrypt(flag_enc), 16).decode()
%20 %20print()
%20 %20print("AES-128%20key%20schedule")
%20 %20print(f"round00(master_key)%20= {round_keys[0].hex()}")
%20 %20print(f"round10_key%20 %20 %20 %20 %20= {round10_key.hex()}")
%20 %20print()
%20 %20print("Final%20decrypt")
%20 %20print("mode%20=%20AES-ECB")
%20 %20print(f"key%20 =%20round10_key")
%20 %20print(f"flag%20= {flag}")
if __name__%20== "__main__":
%20 %20main()
先解析公钥，得到%201024-bit%20RSA，公钥指数为常见的 65537。密文文件大小为 23168 字节，而%201024-bit%20RSA%20的单个密文块长度是 128 字节
23168%20/%20128%20=%20181
所以 encrypted.bin 不是一个单独%20RSA%20密文，而是由%20181%20个%20RSA%20密文块拼接而成。后面恢复私钥后，需要按%20128%20字节分块逐块解密。
Fermat%20分解
设：
n%20=%20p%20*%20q
a%20=%20(p%20+%20q)%20/%202
b%20=%20(q%20-%20p)%20/%202
则：
n%20=%20a^2%20-%20b^2%20=%20(a%20-%20b)(a%20+%20b)
也就是说，从 ceil(sqrt(n)) 开始枚举 a，只要发现：
a^2%20-%20n%20=%20b^2
就可以得到：
p%20=%20a%20-%20b
q%20=%20a%20+%20b
p 和 q 的距离确实很小，Fermat%20分解一次就能命中
p%20=%2012751828551709836069103027639236748210345732901386209604520393543958229433834765694925409378355486164691077551258041029878391501131185664614971880587757401
q%20=%2012751828551709836069103027639236748210345732901386209604520393543958229433834765694925409378355486164691077551258041029878391501131185675868445062344890849
q%20-%20p%20=%2011253473181757133448
恢复 p、q 后，就可以计算：
phi(n)%20=%20(p%20-%201)%20*%20(q%20-%201)
d%20=%20e^{-1}%20mod%20phi(n)
然后构造%20RSA%20私钥。
#!/usr/bin/env%20python3
from __future__ import annotations
import argparse
from math import isqrt
from pathlib import Path
from Crypto.Cipher import PKCS1_OAEP
from Crypto.Hash import SHA1,%20SHA224,%20SHA256,%20SHA384,%20SHA512
from Crypto.PublicKey import RSA
from Crypto.Util.number import inverse
HASH_CANDIDATES%20=%20[
%20 %20("SHA1",%20SHA1),
%20 %20("SHA224",%20SHA224),
%20 %20("SHA256",%20SHA256),
%20 %20("SHA384",%20SHA384),
%20 %20("SHA512",%20SHA512),
]
def fermat_factor(n:%20int) ->%20tuple[int,%20int]:
%20 %20a%20=%20isqrt(n)
%20 if a%20*%20a%20<%20n:
%20 %20 %20 %20a%20+= 1
%20 whileTrue:
%20 %20 %20 %20b2%20=%20a%20*%20a%20-%20n
%20 %20 %20 %20b%20=%20isqrt(b2)
%20 %20 %20 if b%20*%20b%20==%20b2:
%20 %20 %20 %20 %20 %20p%20=%20a%20-%20b
%20 %20 %20 %20 %20 %20q%20=%20a%20+%20b
%20 %20 %20 %20 %20 if p%20*%20q%20==%20n:
%20 %20 %20 %20 %20 %20 %20 return int(p),%20int(q)
%20 %20 %20 %20a%20+= 1
def split_blocks(data:%20bytes,%20block_size:%20int) ->%20list[bytes]:
%20 if len(data)%20%%20block_size%20!= 0:
%20 %20 %20 raise ValueError(
%20 %20 %20 %20 %20 f"ciphertext%20length {len(data)} is%20not%20a%20multiple%20of%20block%20size {block_size}"
%20 %20 %20 %20)
%20 return [data[i%20:%20i%20+%20block_size] for i in range(0,%20len(data),%20block_size)]
def decrypt_oaep_blocks(
%20 %20key:%20RSA.RsaKey,%20blocks:%20list[bytes]
) ->%20tuple[str,%20bytes]%20| None:
%20 for hash_name,%20hash_mod in HASH_CANDIDATES:
%20 %20 %20 %20cipher%20=%20PKCS1_OAEP.new(key,%20hashAlgo=hash_mod)
%20 %20 %20 %20plaintext_parts%20=%20[]
%20 %20 %20 try:
%20 %20 %20 %20 %20 for block in blocks:
%20 %20 %20 %20 %20 %20 %20 %20plaintext_parts.append(cipher.decrypt(block))
%20 %20 %20 except ValueError:
%20 %20 %20 %20 %20 continue
%20 %20 %20 return hash_name, b"".join(plaintext_parts)
%20 returnNone
def guess_extension(blob:%20bytes) ->%20str:
%20 if blob.startswith(b"\x89PNG\r\n\x1a\n"):
%20 %20 %20 return".png"
if blob.startswith(b"\xff\xd8\xff"):
return".jpg"
if blob.startswith(b"GIF87a") or blob.startswith(b"GIF89a"):
return".gif"
if blob.startswith(b"BM"):
return".bmp"
if blob.startswith(b"RIFF") and blob[8:12] == b"WEBP":
return".webp"
return".bin"
def solve(input_dir: Path, output_path: Path | None) -> tuple[Path, dict[str, int | str]]:
input_dir = input_dir.resolve()
pubkey_path = input_dir / "pubkey.pem"
encrypted_path = input_dir / "encrypted.bin"
public_key = RSA.import_key(pubkey_path.read_bytes())
n = public_key.n
e = public_key.e
p, q = fermat_factor(n)
phi = (p - 1) * (q - 1)
d = inverse(e, phi)
private_key = RSA.construct((n, e, d, p, q))
ciphertext = encrypted_path.read_bytes()
blocks = split_blocks(ciphertext, public_key.size_in_bytes())
decrypted = decrypt_oaep_blocks(private_key, blocks)
if decrypted isNone:
raise RuntimeError("unable to decrypt all blocks with common OAEP hash choices")
hash_name, plaintext = decrypted
if output_path isNone:
output_path = input_dir / f"decrypted_image{guess_extension(plaintext)}"
output_path.write_bytes(plaintext)
metadata: dict[str, int | str] = {
"n_bits": public_key.size_in_bits(),
"e": e,
"p": p,
"q": q,
"prime_gap": q - p,
"block_size": public_key.size_in_bytes(),
"block_count": len(blocks),
"oaep_hash": hash_name,
"plaintext_len": len(plaintext),
}
return output_path, metadata
def main() -> None:
parser = argparse.ArgumentParser(description="Solve q4_close_primes")
parser.add_argument(
"--input-dir",
type=Path,
default=Path("."),
help="directory containing pubkey.pem and encrypted.bin; defaults to current directory",
)
parser.add_argument(
"--output",
type=Path,
default=None,
help="output image path; defaults to decrypted_image.<ext> in the input directory",
)
args = parser.parse_args()
output_path, metadata = solve(args.input_dir, args.output)
print(f"[+] saved plaintext to: {output_path}")
for key, value in metadata.items():
print(f"[+] {key}: {value}")
if __name__ == "__main__":
main()
免责声明：
本文所载程序、技术方法仅面向合法合规的安全研究与教学场景，旨在提升网络安全防护能力，具有明确的技术研究属性。
任何单位或个人未经授权，将本文内容用于攻击、破坏等非法用途的，由此引发的全部法律责任、民事赔偿及连带责任，均由行为人独立承担，本站不承担任何连带责任。
本站内容均为技术交流与知识分享目的发布，若存在版权侵权或其他异议，请通过邮件联系处理，具体联系方式可点击页面上方的联系我。
本文转载自：赛查查 《2026第十届御网杯 数据安全赛道WP》
版权声明
本站仅做备份收录，仅供研究与教学参考之用。
读者将信息用于其他用途的，全部法律及连带责任由读者自行承担，本站不承担任何责任。
评论