Coverage for /var/devmt/py/utils4_1.8.0/utils4/srccheck.py: 100%

80 statements  

« prev     ^ index     » next       coverage.py v7.6.9, created at 2025-08-26 21:39 +0100

1#!/usr/bin/env python 

2# -*- coding: utf-8 -*- 

3""" 

4:Purpose: This module is used to perform checksum calculations on a 

5 collection of files to verify if the checksum *calculated* on each 

6 file matches the *expected* checksum value. 

7 

8 In practical terms, an application can call the 

9 :meth:`~SourceCheck.check` method by passing a list of filepaths 

10 to be checksummed, along with a reference file (containing the 

11 expected checksums). If the checksum values match the reference 

12 file, a value of ``True`` is returned to the caller application, 

13 signaling the inspected source code files have *not* been modified 

14 and are 'safe' for use. Otherwise, a value of ``False`` is 

15 returned to the caller the filenames of each failing file are 

16 printed to the terminal. 

17 

18:Developer: J Berendt 

19:Email: development@s3dev.uk 

20 

21:Comments: n/a 

22 

23:Example usage: 

24 

25 Generate an *un-encrypted* reference file:: 

26 

27 >>> from utils4.srccheck import srccheck 

28 

29 >>> files = ['list.c', 'of.py', 'files.sql'] 

30 >>> srccheck.generate(filepaths=files, encrypt=False) 

31 

32 

33 Verify checksums from within an application, with an *un-encrypted* 

34 reference file:: 

35 

36 >>> from utils4.srccheck import srccheck 

37 

38 >>> srccheck.check(ref_file='path/to/srccheck.ref') 

39 True 

40 

41 Generate an **encrypted** reference file:: 

42 

43 >>> from utils4.srccheck import srccheck 

44 

45 >>> files = ['list.c', 'of.py', 'files.sql'] 

46 >>> srccheck.generate(filepaths=files, encrypt=True) 

47 

48 

49 Verify checksums from within an application, with an *encrypted* reference 

50 file:: 

51 

52 >>> from utils4.srccheck import srccheck 

53 

54 >>> srccheck.check(ref_file='path/to/srccheck.ref', 

55 key_file='path/to/srccheck.key') 

56 True 

57 

58 

59 **Advanced usage:** 

60 

61 If you wish to *delay the output* of mismatched files (to give the caller 

62 application display control), the caller can redirected the output from 

63 the :meth:`~SourceCheck.check` method into a buffer and display at a more 

64 appropriate time. For example:: 

65 

66 >>> from contextlib import redirect_stdout 

67 >>> from io import StringIO 

68 >>> from utils4.srccheck import srccheck 

69 

70 >>> buff = StringIO() 

71 >>> with redirect_stdout(buff): 

72 >>> test = srccheck.check(ref_file='path/to/srccheck.ref') 

73 

74 >>> # ... 

75 

76 >>> if not test: 

77 >>> print(buff.getvalue()) 

78 >>> buff.close() 

79 

80 Checksum verification has failed for the following: 

81 - 02-01_first.c 

82 - 10-09_ptr_exchange.c 

83 - 06-ex07.c 

84 - 15-ex05_col_output.c 

85 - 02-03_multi_lines.c 

86 

87""" 

88# pylint: disable=wrong-import-order 

89 

90import json 

91import os 

92import pickle 

93import sys 

94import uuid 

95from cryptography import fernet 

96from typing import List 

97from utils4.crypto import crypto 

98 

99 

100class SourceCheck: 

101 """Verify source code checksums values are as expected.""" 

102 

103 def check(self, ref_file: str, key_file: str='') -> bool: 

104 """Verify the provided source code file checksums are as expected. 

105 

106 If any checksum do not match, the names of those files are reported 

107 to the terminal. 

108 

109 Args: 

110 ref_file (str): Full path to the reference file containing the 

111 full paths to the file(s) to be tested and the associated 

112 checksum value(s). 

113 key_file (str, optional): Full path to the key file. If a key file 

114 is not provided, the method assumes the reference file is in 

115 plaintext CSV and does not attempt to decrypt. 

116 Defaults to ''. 

117 

118 Note: 

119 If the ``key_file`` argument is *not* provided, it is assumed the 

120 ``ref_file`` is a plaintext CSV file, and decryption is *not* 

121 attempted. 

122 

123 If the ``key_file`` argument *is* provided, it is assumed the 

124 ``ref_file`` has been encrypted, and decryption is carried out. 

125 

126 Raises: 

127 FileNotFoundError: If either the reference file, or key file do 

128 not exist. 

129 

130 Returns: 

131 bool: True if all file's checksum values agree with the checksum 

132 listed in the reference file; otherwise False. 

133 

134 """ 

135 # pylint: disable=no-else-return 

136 if not os.path.exists(ref_file): 

137 raise FileNotFoundError(f'Reference file not found: {ref_file}') 

138 if all([key_file, not os.path.exists(key_file)]): 

139 raise FileNotFoundError(f'Key file not found: {key_file}') 

140 if key_file: 

141 # Decrypt reference file. 

142 with open(ref_file, 'rb') as rfp: 

143 data = pickle.load(rfp) 

144 with open(key_file, 'rb') as kfp: 

145 f = fernet.Fernet(kfp.read()) 

146 ref = json.loads(f.decrypt(data).decode()) 

147 else: 

148 # Read plaintext reference file. 

149 ref = {} 

150 with open(ref_file, 'r', encoding='utf-8') as rfp: 

151 for line in rfp: 

152 ref.update([line.strip().split(',')]) 

153 chksums = self._checksum(files=ref.keys()) 

154 # Object check for quick validation. 

155 if chksums == ref: 

156 return True 

157 else: 

158 self._report_mismatches(checksums=chksums, reference=ref) 

159 return False 

160 

161 def generate(self, filepaths: List[str], encrypt: bool=False): 

162 """Generate the reference file containing the source file checksums, 

163 and the associated key file. 

164 

165 Args: 

166 filepaths (list[str]): A list of full paths which are to be 

167 included in the reference file. 

168 encrypt (bool, optional): Encrypt the reference file and generate 

169 a key file. Defaults to False. 

170 

171 :Reference File: 

172 

173 **If unencrypted:** 

174 

175 The reference file is a flat, plaintext CSV file with the file 

176 path as the first field and the checksum value as the second field. 

177 

178 For example:: 

179 

180 filepath_01,md5_hash_string_01 

181 filepath_02,md5_hash_string_02 

182 filepath_03,md5_hash_string_03 

183 ... 

184 filepath_NN,md5_hash_string_NN 

185 

186 **If encrypted:** 

187 

188 The reference file contains is a serialised, encrypted 

189 representation of the full path and associated checksum value for 

190 all provided files, in JSON format. This data is written to the 

191 ``srccheck.ref`` file. 

192 

193 A unique encryption key is created and stored with *each* call to 

194 this method, and stored to the ``srccheck.key`` file. 

195 

196 To perform checks, both the reference file *and* the key file must 

197 be provided to the :meth:`~check` method. 

198 

199 .. note:: These files are a **pair**. If one file is lost, the 

200 other file is useless. 

201 

202 :Layout: 

203 

204 **If encrypted:** 

205 

206 The layout of the *deserialised* and *decrypted* reference file is 

207 in basic JSON format, with the filename as the keys, and checksum 

208 values as the values. 

209 

210 For example:: 

211 

212 {"filepath_01": "md5_hash_string_01", 

213 "filepath_02": "md5_hash_string_02", 

214 "filepath_03": "md5_hash_string_03", 

215 ..., 

216 "filepath_NN": "md5_hash_string_NN"} 

217 

218 Raises: 

219 FileNotFoundError: If any of the files provided to the 

220 ``filepaths`` argument do not exist. 

221 

222 """ 

223 if not self._all_files_exist(files=filepaths): 

224 raise FileNotFoundError('The files listed above were not found.') 

225 op_ref, op_key = self._build_outpaths() 

226 chksums = self._checksum(files=filepaths) 

227 if encrypt: 

228 key = crypto.b64(uuid.uuid4().hex, decode=False) 

229 with open(op_key, 'wb') as kfp: 

230 kfp.write(key) 

231 f = fernet.Fernet(key=key) 

232 with open(op_ref, 'wb') as rfp: 

233 pickle.dump(f.encrypt(json.dumps(chksums).encode()), rfp) 

234 print('\nComplete.\nThe reference and key files are available on your desktop.') 

235 else: 

236 with open(op_ref, 'w', encoding='utf-8') as rfp: 

237 for k, v in chksums.items(): 

238 rfp.write(f'{k},{v}\n') 

239 print('\nComplete.\nThe reference file is available on your desktop.') 

240 

241 @staticmethod 

242 def _all_files_exist(files: list) -> bool: 

243 """Verify all provided files exist. 

244 

245 If any file does not exist, the user is alerted via the terminal and a 

246 ``FileNotFoundError`` exception is raised by the caller. 

247 

248 Args: 

249 files (list): List of files to be tested. 

250 

251 Returns: 

252 bool: True, if all files exist, otherwise False. 

253 

254 """ 

255 # pylint: disable=consider-using-f-string 

256 success = True 

257 nexist = [] 

258 for f in files: 

259 if not os.path.exists(f): 

260 nexist.append(f) 

261 success = False 

262 if nexist: 

263 print('\nThe following files do not exist:') 

264 print(*map(' - {}'.format, nexist), sep='\n') 

265 print('') 

266 return success 

267 

268 @staticmethod 

269 def _build_outpaths() -> tuple: 

270 """Build the output path to the reference and key files. 

271 

272 Returns: 

273 tuple: Full path to the reference and key files as:: 

274 

275 ('fname.ref', 'fname.key') 

276 

277 """ 

278 _os = sys.platform.lower() 

279 fn_ref = 'srccheck.ref' 

280 fn_key = 'srccheck.key' 

281 if 'win' in _os: # pragma nocover 

282 desk = os.path.join(os.environ.get('USERPROFILE'), 'Desktop') 

283 elif 'lin' in _os: 

284 desk = os.path.join(os.environ.get('HOME'), 'Desktop') 

285 else: # pragma nocover 

286 raise NotImplementedError(f'Not a currently supported OS: {_os}') 

287 return os.path.join(desk, fn_ref), os.path.join(desk, fn_key) 

288 

289 @staticmethod 

290 def _checksum(files: list) -> dict: 

291 """Calculate checksum for all passed files. 

292 

293 Args: 

294 files (list): List of full paths against which a checksum is to be 

295 calculated. 

296 

297 Returns: 

298 dict: A dictionary containing the filename and checksum for all 

299 passed files, as:: 

300 

301 {'fname_01': 'checksum_hash_01', 

302 'fname_02': 'checksum_hash_02', 

303 'fname_03': 'checksum_hash_03', 

304 ..., 

305 'fname_NN': 'checksum_hash_NN'} 

306 

307 """ 

308 return {f: crypto.checksum_md5(path=f) for f in files} 

309 

310 @staticmethod 

311 def _report_mismatches(checksums: dict, reference: dict): 

312 """Report the files for which the checksums do not match. 

313 

314 Args: 

315 checksums (dict): A dictionary containing the recently calculated 

316 checksums. 

317 reference (dict): A dictionary containing the *expected* checksums. 

318 

319 """ 

320 # pylint: disable=consider-using-f-string 

321 m = [] 

322 for k, v in reference.items(): 

323 if checksums.get(k) != v: 

324 m.append(os.path.basename(k)) 

325 print('\nChecksum verification has failed for the following:') 

326 print(*map('- {}'.format, m), sep='\n') 

327 print('') 

328 

329 

330srccheck = SourceCheck()