Python查找ADS文件名代码异常，求排查FindFirstStreamW行问题

我尝试用Python编写代码查找NTFS交替数据流（ADS）文件名，但运行后未得到预期结果，尤其在执行代码行：

myhandler = kernel32.FindFirstStreamW (LPSTR(self.filename), 0, byref(file_infos), 0)

时存在问题。以下是我编写的完整代码：

from ctypes import *

import sys, os
kernel32 = windll.kernel32

LPSTR     = c_wchar_p
DWORD     = c_ulong
LONG      = c_ulong
WCHAR     = c_wchar * 164
LONGLONG  = c_longlong

class LARGE_INTEGER_UNION(Structure):
    _fields_ = [
        ("LowPart", DWORD),
        ("HighPart", LONG),]

class LARGE_INTEGER(Union):
    _fields_ = [
        ("large1", LARGE_INTEGER_UNION),
        ("large2", LARGE_INTEGER_UNION),
        ("QuadPart",    LONGLONG),
    ]

class WIN32_FIND_STREAM_DATA(Structure):
    _fields_ = [
        ("StreamSize", LARGE_INTEGER),
        ("cStreamName", WCHAR),
    ]
    '''
    typedef struct _WIN32_FIND_STREAM_DATA {
      LARGE_INTEGER StreamSize;
      WCHAR         cStreamName[MAX_PATH + 36];
    } WIN32_FIND_STREAM_DATA, *PWIN32_FIND_STREAM_DATA;
    ''' 

class ADS():
    def __init__(self, file):
        self.filename = file
        self.streams = self.retrieveStreams()
          
        
    def retrieveStreams(self):
        file_infos = WIN32_FIND_STREAM_DATA()
        streamlist = list()
        myhandler = kernel32.FindFirstStreamW (LPSTR(self.filename), 0, byref(file_infos), 0)
        '''
        HANDLE WINAPI FindFirstStreamW(
          __in        LPCWSTR lpFileName,
          __in        STREAM_INFO_LEVELS InfoLevel, (0 standard, 1 max infos)
          __out       LPVOID lpFindStreamData, (return information about file in a WIN32_FIND_STREAM_DATA if 0 is given in infos_level
          __reserved  DWORD dwFlags (Reserved for future use. This parameter must be zero.) cf: doc
        );
        '''
        
        if not file_infos.cStreamName:
            return streamlist #directories don't have default ADS
        else:
            streamname = file_infos.cStreamName.split(":")[1]
            if streamname: streamlist.append(streamname)
        
            while kernel32.FindNextStreamW(myhandler, byref(file_infos)):
                streamlist.append(file_infos.cStreamName.split(":")[1])

        return streamlist
    
    def getFileName(self):
        return self.filename
    
    def getStreams(self):
        return self.streams
    
    def containStreams(self):
        return len(self.getStreams()) != 0
    
    def addStream(self,newfile):
        #Read in the file content
        if not os.path.exists(newfile):
            return False
        if os.path.exists("%s:%s" % (self.filename, newfile)):
            print("A Stream with the same name already exist.")
            return False
        
        fd = open(newfile, "rb")
        content = fd.read()
        fd.close()
        #Now write it as stream ADS
        fd = open("%s:%s" % (self.filename, newfile), "wb")
        fd.write(content)
        fd.close()
        self.streams.append(newfile)
        return True
    
    def removeStream(self,stream):
        try:
            os.remove("%s:%s" % (self.filename, stream))
            self.streams.remove(stream)
            return True
        except Exception as e:
            return False
    
    def getStreamContent(self,stream):
        fd = open("%s:%s" %(self.filename, stream), "rb")
        content = fd.read()
        fd.close()       
        return content
    
    def extractStream(self, stream):
        fd = open("%s:%s" %(self.filename, stream), "rb")
        content = fd.read()
        fd.close()
        if not os.path.exists(stream):
            fd = open(stream, "wb")
            fd.write(content)
            fd.close()
        else:
            print("File with the same name already exist")
    
    def extractAllStreams(self):
        for s in self.streams:
            self.extractStream(s)    

问题分析与修正方案 #

1. 数据类型定义错误 #

• 宽字符指针命名混淆：LPSTR是ANSI字符串指针，而FindFirstStreamW要求宽字符指针LPCWSTR，应修正为LPCWSTR = c_wchar_p，避免命名误导。
• LARGE_INTEGER结构冗余：原Union定义了两个完全相同的large1和large2，正确结构只需包含一组LowPart+HighPart的结构体和QuadPart。
• cStreamName长度不足：MAX_PATH值为260，cStreamName的标准长度是MAX_PATH + 36，即WCHAR = c_wchar * (260 + 36)，原定义的164长度会导致数据流名称被截断。

2. 函数调用逻辑错误 #

• 未检查调用返回值：FindFirstStreamW返回INVALID_HANDLE_VALUE（值为-1）时表示调用失败，需先判断返回值再处理数据流信息，避免无效内存访问。
• 未释放句柄资源：调用FindFirstStreamW成功后，需在循环结束后调用kernel32.FindClose(myhandler)关闭句柄，防止资源泄漏。
• 冗余参数转换：self.filename可直接传入函数，无需强制转换为LPSTR。

3. 数据流名称处理错误 #

• 默认数据流未过滤：默认数据流名称为:$DATA，需排除该条目，只保留自定义ADS。
• 名称拆分逻辑优化：数据流名称格式为"::<stream_type>"或":<stream_name>:<stream_type>"，应提取:分隔后的第二个部分作为流名称，并排除$DATA。

修正后的代码 #

from ctypes import *
import os

kernel32 = windll.kernel32

# 修正数据类型定义
LPCWSTR = c_wchar_p
DWORD = c_ulong
LONG = c_long
WCHAR = c_wchar * (260 + 36)  # MAX_PATH + 36
LONGLONG = c_longlong

INVALID_HANDLE_VALUE = c_void_p(-1)

class LARGE_INTEGER_UNION(Structure):
    _fields_ = [
        ("LowPart", DWORD),
        ("HighPart", LONG),
    ]

class LARGE_INTEGER(Union):
    _fields_ = [
        ("u", LARGE_INTEGER_UNION),
        ("QuadPart", LONGLONG),
    ]

class WIN32_FIND_STREAM_DATA(Structure):
    _fields_ = [
        ("StreamSize", LARGE_INTEGER),
        ("cStreamName", WCHAR),
    ]

class ADS():
    def __init__(self, file):
        self.filename = file
        self.streams = self.retrieveStreams()
          
    def retrieveStreams(self):
        file_infos = WIN32_FIND_STREAM_DATA()
        streamlist = list()
        
        # 调用FindFirstStreamW并检查返回值
        myhandler = kernel32.FindFirstStreamW(
            LPCWSTR(self.filename),
            0,  # FindStreamInfoStandard
            byref(file_infos),
            0
        )
        
        if myhandler == INVALID_HANDLE_VALUE:
            return streamlist
        
        try:
            # 处理第一个数据流，过滤默认流
            stream_name_parts = file_infos.cStreamName.split(":")
            if len(stream_name_parts) >= 2 and stream_name_parts[1] != "$DATA":
                streamlist.append(stream_name_parts[1])
            
            # 遍历剩余数据流
            while kernel32.FindNextStreamW(myhandler, byref(file_infos)):
                stream_name_parts = file_infos.cStreamName.split(":")
                if len(stream_name_parts) >= 2 and stream_name_parts[1] != "$DATA":
                    streamlist.append(stream_name_parts[1])
        finally:
            # 关闭句柄释放资源
            kernel32.FindClose(myhandler)
        
        return streamlist
    
    def getFileName(self):
        return self.filename
    
    def getStreams(self):
        return self.streams
    
    def containStreams(self):
        return len(self.getStreams()) != 0
    
    def addStream(self, newfile):
        if not os.path.exists(newfile):
            return False
        stream_path = f"{self.filename}:{os.path.basename(newfile)}"
        if os.path.exists(stream_path):
            print("同名数据流已存在")
            return False
        
        with open(newfile, "rb") as f:
            content = f.read()
        
        with open(stream_path, "wb") as f:
            f.write(content)
        
        self.streams.append(os.path.basename(newfile))
        return True
    
    def removeStream(self, stream):
        try:
            os.remove(f"{self.filename}:{stream}")
            self.streams.remove(stream)
            return True
        except Exception:
            return False
    
    def getStreamContent(self, stream):
        with open(f"{self.filename}:{stream}", "rb") as f:
            return f.read()
    
    def extractStream(self, stream):
        content = self.getStreamContent(stream)
        if os.path.exists(stream):
            print("同名文件已存在")
            return
        
        with open(stream, "wb") as f:
            f.write(content)
    
    def extractAllStreams(self):
        for s in self.streams:
            self.extractStream(s)

内容的提问来源于stack exchange，提问作者Osama Aljede