从字节数组解析数据头 Args: data: 字节数组 config: StegaPyConfig配置对象(会被更新) Returns: LSBDataHeader对象
(data, config=None)
| 96 | |
| 97 | @staticmethod |
| 98 | def from_bytes(data, config=None): |
| 99 | """从字节数组解析数据头 |
| 100 | |
| 101 | Args: |
| 102 | data: 字节数组 |
| 103 | config: StegaPyConfig配置对象(会被更新) |
| 104 | |
| 105 | Returns: |
| 106 | LSBDataHeader对象 |
| 107 | """ |
| 108 | if config is None: |
| 109 | from ...config import StegaPyConfig |
| 110 | config = StegaPyConfig() |
| 111 | |
| 112 | offset = 0 |
| 113 | |
| 114 | # 1. 检查DATA_STAMP (9字节) |
| 115 | stamp_len = len(LSBDataHeader.DATA_STAMP) |
| 116 | if len(data) < offset + stamp_len: |
| 117 | raise ValueError("数据头长度不足,无法读取DATA_STAMP") |
| 118 | stamp = data[offset:offset+stamp_len] |
| 119 | if stamp != LSBDataHeader.DATA_STAMP: |
| 120 | raise ValueError(f"无效的数据头标记,期望'STEGAPY ',实际为'{stamp.decode('utf-8', errors='ignore')}'") |
| 121 | offset += stamp_len |
| 122 | |
| 123 | # 2. 检查HEADER_VERSION (1字节) |
| 124 | version_len = len(LSBDataHeader.HEADER_VERSION) |
| 125 | if len(data) < offset + version_len: |
| 126 | raise ValueError("数据头长度不足,无法读取HEADER_VERSION") |
| 127 | version = data[offset:offset+version_len] |
| 128 | if version != LSBDataHeader.HEADER_VERSION: |
| 129 | raise ValueError(f"无效的头版本,期望版本2,实际为{version[0]}") |
| 130 | offset += version_len |
| 131 | |
| 132 | # 3. 读取FIXED_HEADER (8字节) |
| 133 | if len(data) < offset + LSBDataHeader.FIXED_HEADER_LENGTH: |
| 134 | raise ValueError("数据头长度不足,无法读取FIXED_HEADER") |
| 135 | fixed_header = data[offset:offset+LSBDataHeader.FIXED_HEADER_LENGTH] |
| 136 | # dataLength (4字节,小端序) |
| 137 | data_length = struct.unpack('<I', fixed_header[0:4])[0] |
| 138 | # channelBitsUsed (1字节) |
| 139 | channel_bits_used = fixed_header[4] |
| 140 | # fileNameLen (1字节) |
| 141 | filename_len = fixed_header[5] |
| 142 | # useCompression (1字节) |
| 143 | config.set_use_compression(fixed_header[6] == 1) |
| 144 | # useEncryption (1字节) |
| 145 | config.set_use_encryption(fixed_header[7] == 1) |
| 146 | offset += LSBDataHeader.FIXED_HEADER_LENGTH |
| 147 | |
| 148 | # 4. 读取CRYPT_ALGO (8字节) |
| 149 | if len(data) < offset + LSBDataHeader.CRYPT_ALGO_LENGTH: |
| 150 | raise ValueError("数据头长度不足,无法读取CRYPT_ALGO") |
| 151 | crypt_algo = data[offset:offset+LSBDataHeader.CRYPT_ALGO_LENGTH] |
| 152 | # 去除尾部空格 |
| 153 | crypt_algo_str = crypt_algo.rstrip(b' ').decode('utf-8', errors='ignore') |
| 154 | if crypt_algo_str: |
| 155 | config.set_encryption_algorithm(crypt_algo_str) |
no test coverage detected