Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/unblob/handlers/archive/qnap/qnap_nas.py: 48%

1import io 

2 

3import attrs 

4from pyperscan import Flag, Pattern, Scan, StreamDatabase 

5from structlog import get_logger 

6 

7from unblob.file_utils import File, stream_scan 

8from unblob.models import ( 

9 Endian, 

10 Handler, 

11 HandlerDoc, 

12 HandlerType, 

13 HexString, 

14 StructParser, 

15 ValidChunk, 

16) 

17 

18from ._qnap import C_DEFINITIONS, QnapExtractor 

19 

20logger = get_logger() 

21 

22SECRET = "QNAPNASVERSION" # noqa: S105 

23 

24FOOTER_PATTERN = [ 

25 HexString("69 63 70 6e 61 73"), # encrypted gzip stream start bytes 

26] 

27 

28_STRUCT_PARSER = StructParser(C_DEFINITIONS) 

29 

30 

31@attrs.define 

32class QTSSearchContext: 

33 start_offset: int 

34 file: File 

35 end_offset: int 

36 

37 

38def is_valid_header(header) -> bool: 

39 try: 

40 header.device_id.decode("utf-8") 

41 header.file_version.decode("utf-8") 

42 header.firmware_date.decode("utf-8") 

43 header.revision.decode("utf-8") 

44 except UnicodeDecodeError: 

45 return False 

46 return True 

47 

48 

49def _hyperscan_match( 

50 context: QTSSearchContext, pattern_id: int, offset: int, end: int 

51) -> Scan: 

52 del pattern_id, end # unused arguments 

53 if offset < context.start_offset: 

54 return Scan.Continue 

55 context.file.seek(offset, io.SEEK_SET) 

56 header = _STRUCT_PARSER.parse("qnap_header_t", context.file, Endian.LITTLE) 

57 logger.debug("qnap_header_t", header=header) 

58 

59 if is_valid_header(header): 

60 encrypted_data_len = offset - context.start_offset 

61 if header.encrypted_len == 0 or header.encrypted_len > encrypted_data_len: 

62 return Scan.Continue 

63 context.end_offset = context.file.tell() 

64 return Scan.Terminate 

65 return Scan.Continue 

66 

67 

68def build_stream_end_scan_db(pattern_list): 

69 return StreamDatabase( 

70 *(Pattern(p.as_regex(), Flag.SOM_LEFTMOST, Flag.DOTALL) for p in pattern_list) 

71 ) 

72 

73 

74hyperscan_stream_end_magic_db = build_stream_end_scan_db(FOOTER_PATTERN) 

75 

76 

77class QnapNasExtractor(QnapExtractor): 

78 def _get_secret(self, header) -> str: 

79 return SECRET + header.file_version.decode("utf-8")[0] 

80 

81 

82class QnapHandler(Handler): 

83 NAME = "qnap_nas" 

84 

85 PATTERNS = [ 

86 HexString("F5 7B 47 03"), 

87 ] 

88 EXTRACTOR = QnapNasExtractor() 

89 

90 DOC = HandlerDoc( 

91 name="QNAP NAS", 

92 description="QNAP NAS firmware files consist of a custom header, encrypted data sections, and a footer marking the end of the encrypted stream. The header contains metadata such as device ID, firmware version, and encryption details.", 

93 handler_type=HandlerType.ARCHIVE, 

94 vendor="QNAP", 

95 references=[], 

96 limitations=[], 

97 ) 

98 

99 def calculate_chunk(self, file: File, start_offset: int) -> ValidChunk | None: 

100 context = QTSSearchContext(start_offset=start_offset, file=file, end_offset=-1) 

101 

102 try: 

103 scanner = hyperscan_stream_end_magic_db.build(context, _hyperscan_match) # type: ignore 

104 stream_scan(scanner, file) 

105 except Exception as e: 

106 logger.debug( 

107 "Error scanning for QNAP patterns", 

108 error=e, 

109 ) 

110 if context.end_offset > 0: 

111 return ValidChunk(start_offset=start_offset, end_offset=context.end_offset) 

112 return None