glasswall.libraries.archive_manager.archive_manager

  1import ctypes as ct
  2import functools
  3import io
  4import os
  5from typing import Optional, Union
  6
  7import glasswall
  8from glasswall import determine_file_type as dft
  9from glasswall import utils
 10from glasswall.config.logging import log
 11from glasswall.libraries.archive_manager import errors, successes
 12from glasswall.libraries.library import Library
 13
 14
 15class ArchiveManager(Library):
 16    """ A high level Python wrapper for Glasswall Archive Manager. """
 17
 18    def __init__(self, library_path):
 19        super().__init__(library_path)
 20        self.library = self.load_library(os.path.abspath(library_path))
 21
 22        log.info(f"Loaded Glasswall {self.__class__.__name__} version {self.version()} from {self.library_path}")
 23
 24    def version(self):
 25        """ Returns the Glasswall library version.
 26
 27        Returns:
 28            version (str): The Glasswall library version.
 29        """
 30        # API function declaration
 31        self.library.GwArchiveVersion.restype = ct.c_char_p
 32
 33        # API call
 34        version = self.library.GwArchiveVersion()
 35
 36        # Convert to Python string
 37        version = ct.string_at(version).decode()
 38
 39        return version
 40
 41    def release(self):
 42        """ Releases any resources held by the Glasswall Archive Manager library. """
 43        self.library.GwArchiveDone()
 44
 45    @property
 46    @functools.lru_cache()
 47    def supported_archives(self):
 48        """ Returns a list of supported archive file formats. """
 49
 50        # API function declaration
 51        self.library.GwSupportedFiletypes.restype = ct.c_char_p
 52
 53        # API call
 54        result = self.library.GwSupportedFiletypes()  # b'7z,bz2,gz,rar,tar,xz,zip,'
 55
 56        # Convert to Python string
 57        result = ct.string_at(result).decode()  # 7z,bz2,gz,rar,tar,xz,zip,
 58
 59        # Convert comma separated str to list, remove empty trailing element, sort
 60        result = sorted(filter(None, result.split(",")))
 61
 62        return result
 63
 64    @functools.lru_cache()
 65    def is_supported_archive(self, archive_type: str):
 66        """ Returns True if the archive type (e.g. `7z`) is supported. """
 67
 68        # API function declaration
 69        self.library.GwIsSupportedArchiveType.argtypes = [
 70            ct.c_char_p
 71        ]
 72        self.library.GwIsSupportedArchiveType.restype = ct.c_bool
 73
 74        ct_archive_type = ct.c_char_p(archive_type.encode())  # const char* type
 75
 76        result = self.library.GwIsSupportedArchiveType(ct_archive_type)
 77
 78        return result
 79
 80    def list_archive_paths(self, directory: str, recursive: bool = True, absolute: bool = True, followlinks: bool = True):
 81        """ Returns a list of file paths of supported archives in a directory and all of its subdirectories. """
 82        return [
 83            file_path
 84            for file_path in glasswall.utils.list_file_paths(
 85                directory=directory,
 86                recursive=recursive,
 87                absolute=absolute,
 88                followlinks=followlinks,
 89            )
 90            if self.is_supported_archive(self.determine_file_type(file_path, as_string=True, raise_unsupported=False))
 91        ]
 92
 93    def determine_file_type(self, input_file: str, as_string: bool = False, raise_unsupported: bool = True):
 94        """ Returns an int representing the file type of an archive.
 95
 96        Args:
 97            input_file (str) The input file path.
 98            as_string (bool, optional): Return file type as string, eg: "xz" instead of: 262. Defaults to False.
 99            raise_unsupported (bool, optional): Default True. Raise exceptions when Glasswall encounters an error. Fail silently if False.
100
101        Returns:
102            file_type (Union[int, str]): The file format.
103        """
104        if not os.path.isfile(input_file):
105            raise FileNotFoundError(input_file)
106
107        # API function declaration
108        self.library.GwDetermineArchiveTypeFromFile.argtypes = [
109            ct.c_char_p
110        ]
111
112        # Variable initialisation
113        ct_input_file = ct.c_char_p(input_file.encode())  # const char * inputFilePath)
114
115        with utils.CwdHandler(new_cwd=self.library_path):
116            # API call
117            file_type = self.library.GwDetermineArchiveTypeFromFile(
118                ct_input_file
119            )
120
121        file_type_as_string = dft.file_type_int_to_str(file_type)
122        input_file_repr = f"{type(input_file)} length {len(input_file)}" if isinstance(input_file, (bytes, bytearray,)) else input_file.__sizeof__() if isinstance(input_file, io.BytesIO) else input_file
123
124        if not dft.is_success(file_type):
125            if raise_unsupported:
126                log.warning(f"\n\tfile_type: {file_type}\n\tfile_type_as_string: {file_type_as_string}\n\tinput_file: {input_file_repr}")
127                raise dft.int_class_map.get(file_type, dft.errors.UnknownErrorCode)(file_type)
128            else:
129                log.debug(f"\n\tfile_type: {file_type}\n\tfile_type_as_string: {file_type_as_string}\n\tinput_file: {input_file_repr}")
130        else:
131            log.debug(f"\n\tfile_type: {file_type}\n\tfile_type_as_string: {file_type_as_string}\n\tinput_file: {input_file_repr}")
132
133        if as_string:
134            return file_type_as_string
135
136        return file_type
137
138    def analyse_archive(self, input_file: Union[str, bytes, bytearray, io.BytesIO], output_file: Optional[str] = None, output_report: Optional[str] = None, content_management_policy: Union[None, str, bytes, bytearray, io.BytesIO, glasswall.content_management.policies.ArchiveManager] = None, raise_unsupported: bool = True):
139        """ Extracts the input_file archive and processes each file within the archive using the Glasswall engine. Repackages all files regenerated by the Glasswall engine into a new archive, optionally writing the new archive and report to the paths specified by output_file and output_report.
140
141        Args:
142            input_file (Union[str, bytes, bytearray, io.BytesIO]): The archive file path or bytes.
143            output_file (Optional[str], optional): Default None. If str, write the archive to the output_file path.
144            output_report (Optional[str], optional): Default None. If str, write the analysis report to the output_report path.
145            content_management_policy (Union[None, str, bytes, bytearray, io.BytesIO, glasswall.content_management.policies.ArchiveManager], optional): The content management policy to apply.
146            raise_unsupported (bool, optional): Default True. Raise exceptions when Glasswall encounters an error. Fail silently if False.
147
148        Returns:
149            gw_return_object (glasswall.GwReturnObj): An instance of class glasswall.GwReturnObj containing attributes including: "status" (int), "output_file" (bytes), "output_report" (bytes)
150        """
151        # Validate arg types
152        if not isinstance(input_file, (str, bytes, bytearray, io.BytesIO)):
153            raise TypeError(input_file)
154        if not isinstance(output_file, (type(None), str)):
155            raise TypeError(output_file)
156        if not isinstance(output_report, (type(None), str)):
157            raise TypeError(output_report)
158        if not isinstance(content_management_policy, (type(None), str, bytes, bytearray, io.BytesIO, glasswall.content_management.policies.policy.Policy)):
159            raise TypeError(content_management_policy)
160
161        # Convert string path arguments to absolute paths
162        if isinstance(input_file, str):
163            input_file = os.path.abspath(input_file)
164        if isinstance(output_file, str):
165            output_file = os.path.abspath(output_file)
166        if isinstance(output_report, str):
167            output_report = os.path.abspath(output_report)
168
169        # Convert inputs to bytes
170        if isinstance(input_file, str):
171            if not os.path.isfile(input_file):
172                raise FileNotFoundError(input_file)
173            with open(input_file, "rb") as f:
174                input_file_bytes = f.read()
175        elif isinstance(input_file, (bytes, bytearray, io.BytesIO)):
176            input_file_bytes = utils.as_bytes(input_file)
177
178        if isinstance(content_management_policy, str) and os.path.isfile(content_management_policy):
179            with open(content_management_policy, "rb") as f:
180                content_management_policy = f.read()
181        elif isinstance(content_management_policy, type(None)):
182            # Load default
183            content_management_policy = glasswall.content_management.policies.ArchiveManager(default="sanitise", default_archive_manager="process")
184        content_management_policy = utils.validate_xml(content_management_policy)
185
186        # API function declaration
187        self.library.GwFileAnalysisArchive.argtypes = [
188            ct.c_void_p,  # void *inputBuffer
189            ct.c_size_t,  # size_t inputBufferLength
190            ct.POINTER(ct.c_void_p),  # void **outputFileBuffer
191            ct.POINTER(ct.c_size_t),  # size_t *outputFileBufferLength
192            ct.POINTER(ct.c_void_p),  # void **outputAnalysisReportBuffer
193            ct.POINTER(ct.c_size_t),  # size_t *outputAnalysisReportBufferLength
194            ct.c_char_p  # const char *xmlConfigString
195        ]
196
197        # Variable initialisation
198        gw_return_object = glasswall.GwReturnObj()
199        gw_return_object.input_buffer = ct.create_string_buffer(input_file_bytes)
200        gw_return_object.input_buffer_length = ct.c_size_t(len(input_file_bytes))
201        gw_return_object.output_buffer = ct.c_void_p()
202        gw_return_object.output_buffer_length = ct.c_size_t()
203        gw_return_object.output_report_buffer = ct.c_void_p()
204        gw_return_object.output_report_buffer_length = ct.c_size_t()
205        gw_return_object.content_management_policy = ct.c_char_p(content_management_policy.encode())
206
207        with utils.CwdHandler(new_cwd=self.library_path):
208            # API call
209            gw_return_object.status = self.library.GwFileAnalysisArchive(
210                gw_return_object.input_buffer,
211                gw_return_object.input_buffer_length,
212                ct.byref(gw_return_object.output_buffer),
213                ct.byref(gw_return_object.output_buffer_length),
214                ct.byref(gw_return_object.output_report_buffer),
215                ct.byref(gw_return_object.output_report_buffer_length),
216                gw_return_object.content_management_policy
217            )
218
219        if gw_return_object.output_buffer and gw_return_object.output_buffer_length:
220            gw_return_object.output_file = utils.buffer_to_bytes(
221                gw_return_object.output_buffer,
222                gw_return_object.output_buffer_length
223            )
224        if gw_return_object.output_report_buffer and gw_return_object.output_report_buffer_length:
225            gw_return_object.output_report = utils.buffer_to_bytes(
226                gw_return_object.output_report_buffer,
227                gw_return_object.output_report_buffer_length
228            )
229
230        # Write output file
231        if hasattr(gw_return_object, "output_file"):
232            if isinstance(output_file, str):
233                os.makedirs(os.path.dirname(output_file), exist_ok=True)
234                with open(output_file, "wb") as f:
235                    f.write(gw_return_object.output_file)
236
237        # Write output report
238        if hasattr(gw_return_object, "output_report"):
239            if isinstance(output_report, str):
240                os.makedirs(os.path.dirname(output_report), exist_ok=True)
241                with open(output_report, "wb") as f:
242                    f.write(gw_return_object.output_report)
243
244        input_file_repr = f"{type(input_file)} length {len(input_file)}" if isinstance(input_file, (bytes, bytearray,)) else input_file.__sizeof__() if isinstance(input_file, io.BytesIO) else input_file
245        if gw_return_object.status not in successes.success_codes:
246            log.error(f"\n\tinput_file: {input_file_repr}\n\toutput_file: {output_file}\n\tstatus: {gw_return_object.status}")
247            if raise_unsupported:
248                raise errors.error_codes.get(gw_return_object.status, errors.UnknownErrorCode)(gw_return_object.status)
249        else:
250            log.debug(f"\n\tinput_file: {input_file_repr}\n\toutput_file: {output_file}\n\tstatus: {gw_return_object.status}")
251
252        self.release()
253
254        return gw_return_object
255
256    def analyse_directory(self, input_directory: str, output_directory: Optional[str] = None, output_report_directory: Optional[str] = None, content_management_policy: Union[None, str, bytes, bytearray, io.BytesIO, glasswall.content_management.policies.ArchiveManager] = None, raise_unsupported: bool = True):
257        """ Calls analyse_archive on each file in input_directory using the given content management configuration. The resulting archives and analysis reports are written to output_directory maintaining the same directory structure as input_directory.
258
259        Args:
260            input_directory (str): The input directory containing archives to analyse.
261            output_directory (Optional[str], optional): Default None. If str, the output directory where the archives containing analysis reports of each file will be written.
262            output_report_directory (Optional[str], optional): Default None. If str, the output directory where xml reports for each archive will be written.
263            content_management_policy (Union[None, str, bytes, bytearray, io.BytesIO, glasswall.content_management.policies.ArchiveManager], optional): The content management policy to apply.
264            raise_unsupported (bool, optional): Default True. Raise exceptions when Glasswall encounters an error. Fail silently if False.
265
266        Returns:
267            analysed_archives_dict (dict): A dictionary of file paths relative to input_directory, and glasswall.GwReturnObj with attributes: "status" (int), "output_file" (bytes), "output_report" (bytes)
268        """
269        analysed_archives_dict = {}
270        # Call analyse_archive on each file in input_directory
271        for input_file in utils.list_file_paths(input_directory):
272            relative_path = os.path.relpath(input_file, input_directory)
273            # Construct paths for output file and output report
274            output_file = None if output_directory is None else os.path.join(os.path.abspath(output_directory), relative_path)
275            output_report = None if output_report_directory is None else os.path.join(os.path.abspath(output_report_directory), relative_path + ".xml")
276
277            result = self.analyse_archive(
278                input_file=input_file,
279                output_file=output_file,
280                output_report=output_report,
281                content_management_policy=content_management_policy,
282                raise_unsupported=raise_unsupported,
283            )
284
285            analysed_archives_dict[relative_path] = result
286
287        return analysed_archives_dict
288
289    def protect_archive(self, input_file: Union[str, bytes, bytearray, io.BytesIO], output_file: Optional[str] = None, output_report: Optional[str] = None, content_management_policy: Union[None, str, bytes, bytearray, io.BytesIO, glasswall.content_management.policies.ArchiveManager] = None, raise_unsupported: bool = True):
290        """ Extracts the input_file archive and processes each file within the archive using the Glasswall engine. Repackages all files regenerated by the Glasswall engine into a new archive, optionally writing the new archive and report to the paths specified by output_file and output_report.
291
292        Args:
293            input_file (Union[str, bytes, bytearray, io.BytesIO]): The archive file path or bytes.
294            output_file (Optional[str], optional): Default None. If str, write the archive to the output_file path.
295            output_report (Optional[str], optional): Default None. If str, write the analysis report to the output_report path.
296            content_management_policy (Union[None, str, bytes, bytearray, io.BytesIO, glasswall.content_management.policies.ArchiveManager], optional): The content management policy to apply.
297            raise_unsupported (bool, optional): Default True. Raise exceptions when Glasswall encounters an error. Fail silently if False.
298
299        Returns:
300            gw_return_object (glasswall.GwReturnObj): An instance of class glasswall.GwReturnObj containing attributes including: "status" (int), "output_file" (bytes), "output_report" (bytes)
301        """
302        # Validate arg types
303        if not isinstance(input_file, (str, bytes, bytearray, io.BytesIO)):
304            raise TypeError(input_file)
305        if not isinstance(output_file, (type(None), str)):
306            raise TypeError(output_file)
307        if not isinstance(output_report, (type(None), str)):
308            raise TypeError(output_report)
309        if not isinstance(content_management_policy, (type(None), str, bytes, bytearray, io.BytesIO, glasswall.content_management.policies.policy.Policy)):
310            raise TypeError(content_management_policy)
311
312        # Convert string path arguments to absolute paths
313        if isinstance(input_file, str):
314            input_file = os.path.abspath(input_file)
315        if isinstance(output_file, str):
316            output_file = os.path.abspath(output_file)
317        if isinstance(output_report, str):
318            output_report = os.path.abspath(output_report)
319
320        # Convert inputs to bytes
321        if isinstance(input_file, str):
322            if not os.path.isfile(input_file):
323                raise FileNotFoundError(input_file)
324            with open(input_file, "rb") as f:
325                input_file_bytes = f.read()
326        elif isinstance(input_file, (bytes, bytearray, io.BytesIO)):
327            input_file_bytes = utils.as_bytes(input_file)
328
329        if isinstance(content_management_policy, str) and os.path.isfile(content_management_policy):
330            with open(content_management_policy, "rb") as f:
331                content_management_policy = f.read()
332        elif isinstance(content_management_policy, type(None)):
333            # Load default
334            content_management_policy = glasswall.content_management.policies.ArchiveManager(default="sanitise", default_archive_manager="process")
335        content_management_policy = utils.validate_xml(content_management_policy)
336
337        # API function declaration
338        self.library.GwFileProtectAndReportArchive.argtypes = [
339            ct.c_void_p,  # void *inputBuffer
340            ct.c_size_t,  # size_t inputBufferLength
341            ct.POINTER(ct.c_void_p),  # void **outputFileBuffer
342            ct.POINTER(ct.c_size_t),  # size_t *outputFileBufferLength
343            ct.POINTER(ct.c_void_p),  # void **outputReportBuffer
344            ct.POINTER(ct.c_size_t),  # size_t *outputReportBufferLength
345            ct.c_char_p  # const char *xmlConfigString
346        ]
347        # Variable initialisation
348        gw_return_object = glasswall.GwReturnObj()
349        gw_return_object.input_buffer = ct.create_string_buffer(input_file_bytes)
350        gw_return_object.input_buffer_length = ct.c_size_t(len(input_file_bytes))
351        gw_return_object.output_buffer = ct.c_void_p()
352        gw_return_object.output_buffer_length = ct.c_size_t()
353        gw_return_object.output_report_buffer = ct.c_void_p()
354        gw_return_object.output_report_buffer_length = ct.c_size_t()
355        gw_return_object.content_management_policy = ct.c_char_p(content_management_policy.encode())
356
357        with utils.CwdHandler(new_cwd=self.library_path):
358            # API call
359            gw_return_object.status = self.library.GwFileProtectAndReportArchive(
360                ct.byref(gw_return_object.input_buffer),
361                gw_return_object.input_buffer_length,
362                ct.byref(gw_return_object.output_buffer),
363                ct.byref(gw_return_object.output_buffer_length),
364                ct.byref(gw_return_object.output_report_buffer),
365                ct.byref(gw_return_object.output_report_buffer_length),
366                gw_return_object.content_management_policy
367            )
368
369        if gw_return_object.output_buffer and gw_return_object.output_buffer_length:
370            gw_return_object.output_file = utils.buffer_to_bytes(
371                gw_return_object.output_buffer,
372                gw_return_object.output_buffer_length
373            )
374        if gw_return_object.output_report_buffer and gw_return_object.output_report_buffer_length:
375            gw_return_object.output_report = utils.buffer_to_bytes(
376                gw_return_object.output_report_buffer,
377                gw_return_object.output_report_buffer_length
378            )
379
380        # Write output file
381        if hasattr(gw_return_object, "output_file"):
382            if isinstance(output_file, str):
383                os.makedirs(os.path.dirname(output_file), exist_ok=True)
384                with open(output_file, "wb") as f:
385                    f.write(gw_return_object.output_file)
386
387        # Write output report
388        if hasattr(gw_return_object, "output_report"):
389            if isinstance(output_report, str):
390                os.makedirs(os.path.dirname(output_report), exist_ok=True)
391                with open(output_report, "wb") as f:
392                    f.write(gw_return_object.output_report)
393
394        input_file_repr = f"{type(input_file)} length {len(input_file)}" if isinstance(input_file, (bytes, bytearray,)) else input_file.__sizeof__() if isinstance(input_file, io.BytesIO) else input_file
395        if gw_return_object.status not in successes.success_codes:
396            log.error(f"\n\tinput_file: {input_file_repr}\n\toutput_file: {output_file}\n\tstatus: {gw_return_object.status}")
397            if raise_unsupported:
398                raise errors.error_codes.get(gw_return_object.status, errors.UnknownErrorCode)(gw_return_object.status)
399        else:
400            log.debug(f"\n\tinput_file: {input_file_repr}\n\toutput_file: {output_file}\n\tstatus: {gw_return_object.status}")
401
402        self.release()
403
404        return gw_return_object
405
406    def protect_directory(self, input_directory: str, output_directory: Optional[str] = None, output_report_directory: Optional[str] = None, content_management_policy: Union[None, str, bytes, bytearray, io.BytesIO, glasswall.content_management.policies.ArchiveManager] = None, raise_unsupported: bool = True):
407        """ Calls protect_archive on each file in input_directory using the given content management configuration. The resulting archives are written to output_directory maintaining the same directory structure as input_directory.
408
409        Args:
410            input_directory (str): The input directory containing archives to protect.
411            output_directory (Optional[str], optional): Default None. If str, the output directory where the archives will be written.
412            output_report_directory (Optional[str], optional): Default None. If str, the output directory where xml reports for each archive will be written.
413            content_management_policy (Union[None, str, bytes, bytearray, io.BytesIO, glasswall.content_management.policies.ArchiveManager], optional): The content management policy to apply.
414            raise_unsupported (bool, optional): Default True. Raise exceptions when Glasswall encounters an error. Fail silently if False.
415
416        Returns:
417            protected_archives_dict (dict): A dictionary of file paths relative to input_directory, and glasswall.GwReturnObj with attributes: "status" (int), "output_file" (bytes), "output_report" (bytes)
418        """
419        protected_archives_dict = {}
420        # Call protect_archive on each file in input_directory to output_directory
421        for input_file in utils.list_file_paths(input_directory):
422            relative_path = os.path.relpath(input_file, input_directory)
423            # Construct paths for output file and output report
424            output_file = None if output_directory is None else os.path.join(os.path.abspath(output_directory), relative_path)
425            output_report = None if output_report_directory is None else os.path.join(os.path.abspath(output_report_directory), relative_path + ".xml")
426
427            result = self.protect_archive(
428                input_file=input_file,
429                output_file=output_file,
430                output_report=output_report,
431                content_management_policy=content_management_policy,
432                raise_unsupported=raise_unsupported,
433            )
434
435            protected_archives_dict[relative_path] = result
436
437        return protected_archives_dict
438
439    def file_to_file_unpack(self, input_file: str, output_directory: str, raise_unsupported: bool = True):
440        # Validate arg types
441        if not isinstance(input_file, str):
442            raise TypeError(input_file)
443        elif not os.path.isfile(input_file):
444            raise FileNotFoundError(input_file)
445        if not isinstance(output_directory, str):
446            raise TypeError(output_directory)
447
448        # API function declaration
449        self.library.GwFileToFileUnpack.argtypes = [
450            ct.c_char_p,
451            ct.c_char_p,
452        ]
453
454        # Variable initialisation
455        gw_return_object = glasswall.GwReturnObj()
456        gw_return_object.ct_input_file = ct.c_char_p(input_file.encode())  # const char* inputFilePath
457        gw_return_object.ct_output_directory = ct.c_char_p(output_directory.encode())  # const char* outputDirPath
458
459        with utils.CwdHandler(new_cwd=self.library_path):
460            # API call
461            gw_return_object.status = self.library.GwFileToFileUnpack(
462                gw_return_object.ct_input_file,
463                gw_return_object.ct_output_directory,
464            )
465
466        if gw_return_object.status not in successes.success_codes:
467            log.error(f"\n\tinput_file: {input_file}\n\tstatus: {gw_return_object.status}")
468            if raise_unsupported:
469                raise errors.error_codes.get(gw_return_object.status, errors.UnknownErrorCode)(gw_return_object.status)
470        else:
471            log.debug(f"\n\tinput_file: {input_file}\n\tstatus: {gw_return_object.status}")
472
473        self.release()
474
475        return gw_return_object
476
477    def file_to_file_pack(self, input_directory: str, output_directory: str, file_type: Optional[str] = None, add_extension: Optional[bool] = True, raise_unsupported: Optional[bool] = True):
478        # Validate arg types
479        if not isinstance(input_directory, str):
480            raise TypeError(input_directory)
481        elif not os.path.isdir(input_directory):
482            raise NotADirectoryError(input_directory)
483        if not isinstance(output_directory, str):
484            raise TypeError(output_directory)
485        if not file_type:
486            file_type = utils.get_file_type(input_directory)
487
488        # Ensure output_directory exists
489        os.makedirs(output_directory, exist_ok=True)
490
491        # API function declaration
492        self.library.GwFileToFilePack.argtypes = [
493            ct.c_char_p,
494            ct.c_char_p,
495            ct.c_char_p,
496            ct.c_int,
497        ]
498
499        # Variable initialisation
500        gw_return_object = glasswall.GwReturnObj()
501        gw_return_object.ct_input_directory = ct.c_char_p(input_directory.encode())  # const char* inputDirPath
502        gw_return_object.ct_output_directory = ct.c_char_p(output_directory.encode())  # const char* outputDirPath
503        gw_return_object.ct_file_type = ct.c_char_p(file_type.encode())  # const char *fileType
504        gw_return_object.ct_add_extension = ct.c_int(int(add_extension))  # int addExtension
505
506        with utils.CwdHandler(new_cwd=self.library_path):
507            # API call
508            gw_return_object.status = self.library.GwFileToFilePack(
509                gw_return_object.ct_input_directory,
510                gw_return_object.ct_output_directory,
511                gw_return_object.ct_file_type,
512                gw_return_object.ct_add_extension,
513            )
514
515        if gw_return_object.status not in successes.success_codes:
516            log.error(f"\n\tinput_directory: {input_directory}\n\tstatus: {gw_return_object.status}")
517            if raise_unsupported:
518                raise errors.error_codes.get(gw_return_object.status, errors.UnknownErrorCode)(gw_return_object.status)
519        else:
520            log.debug(f"\n\tinput_directory: {input_directory}\n\tstatus: {gw_return_object.status}")
521
522        self.release()
523
524        return gw_return_object
525
526    def unpack(self, input_file: str, output_directory: str, recursive: bool = True, include_file_type: bool = False, raise_unsupported: bool = True, delete_origin: bool = False):
527        """ Unpack an archive, maintaining directory structure. Supported archive formats are: "7z", "bz2", "gz", "rar", "tar", "xz", "zip".
528
529        Args:
530            input_file (str): The archive file path
531            output_directory (str): The output directory where the archive will be unpacked to a new directory.
532            recursive (bool, optional): Default True. Recursively unpack all nested archives.
533            include_file_type (bool, optional): Default False. Include the archive format in the directory name. Useful when there are multiple same-named archives of different formats.
534            raise_unsupported (bool, optional): Default True. Raise exceptions when Glasswall encounters an error. Fail silently if False.
535            delete_origin (bool, optional): Default False. Delete input_file after unpacking to output_directory.
536        """
537        # Convert to absolute paths
538        input_file = os.path.abspath(input_file)
539        output_directory = os.path.abspath(output_directory)
540
541        if include_file_type:
542            archive_name = os.path.basename(input_file)
543        else:
544            archive_name = os.path.splitext(os.path.basename(input_file))[0]
545        archive_output_directory = os.path.join(output_directory, archive_name)
546
547        # Unpack
548        log.debug(f"Unpacking\n\tsrc: {input_file}\n\tdst: {archive_output_directory}")
549        result = self.file_to_file_unpack(input_file=input_file, output_directory=archive_output_directory, raise_unsupported=raise_unsupported)
550        if result:
551            status = result.status
552        else:
553            status = None
554
555        if status not in successes.success_codes:
556            log.error(f"\n\tinput_file: {input_file}\n\tstatus: {status}")
557            if raise_unsupported:
558                raise errors.error_codes.get(status, errors.UnknownErrorCode)(status)
559        else:
560            log.debug(f"\n\tinput_file: {input_file}\n\tstatus: {status}")
561
562        if delete_origin:
563            os.remove(input_file)
564
565        if recursive:
566            # Unpack sub archives
567            for subarchive in self.list_archive_paths(archive_output_directory):
568                self.unpack(
569                    input_file=subarchive,
570                    output_directory=archive_output_directory,
571                    recursive=recursive,
572                    raise_unsupported=raise_unsupported,
573                    delete_origin=True
574                )
575
576        return status
577
578    def unpack_directory(self, input_directory: str, output_directory: str, recursive: bool = True, include_file_type: Optional[bool] = False, raise_unsupported: bool = True, delete_origin: bool = False):
579        """ Unpack a directory of archives, maintaining directory structure.
580
581        Args:
582            input_directory (str): The input directory containing archives to unpack.
583            output_directory (str): The output directory where archives will be unpacked to a new directory.
584            recursive (bool, optional): Default True. Recursively unpack all nested archives.
585            include_file_type (bool, optional): Default False. Include the archive format in the directory name. Useful when there are multiple same-named archives of different formats.
586            raise_unsupported (bool, optional): Default True. Raise exceptions when Glasswall encounters an error. Fail silently if False.
587            delete_origin (bool, optional): Default False. Delete input_file after unpacking to output_directory.
588        """
589        # Convert to absolute paths
590        input_directory = os.path.abspath(input_directory)
591        output_directory = os.path.abspath(output_directory)
592
593        for archive_input_file in self.list_archive_paths(input_directory):
594            relative_path = os.path.relpath(archive_input_file, input_directory)
595            archive_output_file = os.path.dirname(os.path.join(output_directory, relative_path))
596            self.unpack(
597                input_file=archive_input_file,
598                output_directory=archive_output_file,
599                recursive=recursive,
600                include_file_type=include_file_type,
601                raise_unsupported=raise_unsupported,
602                delete_origin=delete_origin
603            )
604
605    def pack_directory(self, input_directory: str, output_directory: str, file_type: str, add_extension: Optional[bool] = True, raise_unsupported: Optional[bool] = True, delete_origin: Optional[bool] = False):
606        """ Pack a directory. Supported archive formats are: "7z", "bz2", "gz", "rar", "tar", "xz", "zip".
607
608        Args:
609            input_directory (str): The input directory containing files to archive.
610            output_directory (str): The output directory to store the created archive.
611            file_type (str): The archive file type.
612            add_extension (bool, optional): Default: True. Archive file type extension to result file.
613            raise_unsupported (bool, optional): Default True. Raise exceptions when Glasswall encounters an error. Fail silently if False.
614            delete_origin (bool, optional): Default False. Delete input_directory after packing to output_directory.
615        """
616        # Convert to absolute paths
617        input_directory = os.path.abspath(input_directory)
618        output_directory = os.path.abspath(output_directory)
619
620        # Pack
621        log.debug(f"Packing\n\tsrc: {input_directory}\n\tdst: {output_directory}")
622        status = self.file_to_file_pack(input_directory=input_directory, output_directory=output_directory, file_type=file_type, add_extension=add_extension, raise_unsupported=raise_unsupported).status
623
624        if status not in successes.success_codes:
625            log.error(f"\n\tinput_directory: {input_directory}\n\tstatus: {status}")
626            if raise_unsupported:
627                raise errors.error_codes.get(status, errors.UnknownErrorCode)(status)
628        else:
629            log.debug(f"\n\tinput_directory: {input_directory}\n\tstatus: {status}")
630
631        if delete_origin:
632            utils.delete_directory(input_directory)
633
634        return status
635
636    def export_archive(self, input_file: Union[str, bytes, bytearray, io.BytesIO], output_file: Optional[str] = None, output_report: Optional[str] = None, content_management_policy: Union[None, str, bytes, bytearray, io.BytesIO, glasswall.content_management.policies.ArchiveManager] = None, raise_unsupported: bool = True):
637        """ Exports an archive using the Glasswall engine.
638
639        Args:
640            input_file (Union[str, bytes, bytearray, io.BytesIO]): The archive file path or bytes.
641            output_file (Optional[str], optional): Default None. If str, write the archive to the output_file path.
642            output_report (Optional[str], optional): Default None. If str, write the analysis report to the output_report path.
643            content_management_policy (Union[None, str, bytes, bytearray, io.BytesIO, glasswall.content_management.policies.ArchiveManager], optional): The content management policy to apply.
644            raise_unsupported (bool, optional): Default True. Raise exceptions when Glasswall encounters an error. Fail silently if False.
645
646        Returns:
647            gw_return_object (glasswall.GwReturnObj): An instance of class glasswall.GwReturnObj containing attributes including: "status" (int), "output_file" (bytes), "output_report" (bytes)
648        """
649        # Validate arg types
650        if not isinstance(input_file, (str, bytes, bytearray, io.BytesIO)):
651            raise TypeError(input_file)
652        if not isinstance(output_file, (type(None), str)):
653            raise TypeError(output_file)
654        if not isinstance(output_report, (type(None), str)):
655            raise TypeError(output_report)
656        if not isinstance(content_management_policy, (type(None), str, bytes, bytearray, io.BytesIO, glasswall.content_management.policies.policy.Policy)):
657            raise TypeError(content_management_policy)
658
659        # Convert string path arguments to absolute paths
660        if isinstance(input_file, str):
661            input_file = os.path.abspath(input_file)
662        if isinstance(output_file, str):
663            output_file = os.path.abspath(output_file)
664        if isinstance(output_report, str):
665            output_report = os.path.abspath(output_report)
666
667        # Convert inputs to bytes
668        if isinstance(input_file, str):
669            if not os.path.isfile(input_file):
670                raise FileNotFoundError(input_file)
671            with open(input_file, "rb") as f:
672                input_file_bytes = f.read()
673        elif isinstance(input_file, (bytes, bytearray, io.BytesIO)):
674            input_file_bytes = utils.as_bytes(input_file)
675
676        if isinstance(content_management_policy, str) and os.path.isfile(content_management_policy):
677            with open(content_management_policy, "rb") as f:
678                content_management_policy = f.read()
679        elif isinstance(content_management_policy, type(None)):
680            # Load default
681            content_management_policy = glasswall.content_management.policies.ArchiveManager(default="sanitise", default_archive_manager="process")
682        content_management_policy = utils.validate_xml(content_management_policy)
683
684        # API function declaration
685        self.library.GwFileExportArchive.argtypes = [
686            ct.c_void_p,  # void *inputBuffer
687            ct.c_size_t,  # size_t inputBufferLength
688            ct.POINTER(ct.c_void_p),  # void **outputFileBuffer
689            ct.POINTER(ct.c_size_t),  # size_t *outputFileBufferLength
690            ct.POINTER(ct.c_void_p),  # void **outputReportBuffer
691            ct.POINTER(ct.c_size_t),  # size_t *outputReportBufferLength
692            ct.c_char_p  # const char *xmlConfigString
693        ]
694
695        # Variable initialisation
696        gw_return_object = glasswall.GwReturnObj()
697        gw_return_object.input_buffer = ct.create_string_buffer(input_file_bytes)
698        gw_return_object.input_buffer_length = ct.c_size_t(len(input_file_bytes))
699        gw_return_object.output_buffer = ct.c_void_p()
700        gw_return_object.output_buffer_length = ct.c_size_t()
701        gw_return_object.output_report_buffer = ct.c_void_p()
702        gw_return_object.output_report_buffer_length = ct.c_size_t()
703        gw_return_object.content_management_policy = ct.c_char_p(content_management_policy.encode())
704
705        with utils.CwdHandler(new_cwd=self.library_path):
706            # API call
707            gw_return_object.status = self.library.GwFileExportArchive(
708                gw_return_object.input_buffer,
709                gw_return_object.input_buffer_length,
710                ct.byref(gw_return_object.output_buffer),
711                ct.byref(gw_return_object.output_buffer_length),
712                ct.byref(gw_return_object.output_report_buffer),
713                ct.byref(gw_return_object.output_report_buffer_length),
714                gw_return_object.content_management_policy
715            )
716
717        if gw_return_object.output_buffer and gw_return_object.output_buffer_length:
718            gw_return_object.output_file = utils.buffer_to_bytes(
719                gw_return_object.output_buffer,
720                gw_return_object.output_buffer_length
721            )
722        if gw_return_object.output_report_buffer and gw_return_object.output_report_buffer_length:
723            gw_return_object.output_report = utils.buffer_to_bytes(
724                gw_return_object.output_report_buffer,
725                gw_return_object.output_report_buffer_length
726            )
727
728        # Write output file
729        if hasattr(gw_return_object, "output_file"):
730            if isinstance(output_file, str):
731                os.makedirs(os.path.dirname(output_file), exist_ok=True)
732                with open(output_file, "wb") as f:
733                    f.write(gw_return_object.output_file)
734
735        # Write output report
736        if hasattr(gw_return_object, "output_report"):
737            if isinstance(output_report, str):
738                os.makedirs(os.path.dirname(output_report), exist_ok=True)
739                with open(output_report, "wb") as f:
740                    f.write(gw_return_object.output_report)
741
742        input_file_repr = f"{type(input_file)} length {len(input_file)}" if isinstance(input_file, (bytes, bytearray,)) else input_file.__sizeof__() if isinstance(input_file, io.BytesIO) else input_file
743        if gw_return_object.status not in successes.success_codes:
744            log.error(f"\n\tinput_file: {input_file_repr}\n\toutput_file: {output_file}\n\tstatus: {gw_return_object.status}")
745            if raise_unsupported:
746                raise errors.error_codes.get(gw_return_object.status, errors.UnknownErrorCode)(gw_return_object.status)
747        else:
748            log.debug(f"\n\tinput_file: {input_file_repr}\n\toutput_file: {output_file}\n\tstatus: {gw_return_object.status}")
749
750        self.release()
751
752        return gw_return_object
753
754    def export_directory(self, input_directory: str, output_directory: Optional[str], output_report_directory: Optional[str] = None, content_management_policy: Union[None, str, bytes, bytearray, io.BytesIO, glasswall.content_management.policies.ArchiveManager] = None, raise_unsupported: bool = True):
755        """ Calls export_archive on each file in input_directory. The exported archives are written to output_directory maintaining the same directory structure as input_directory.
756
757        Args:
758            input_directory (str): The input directory containing archives to export.
759            output_directory (Optional[str], optional): Default None. If str, the output directory where the archives will be written.
760            output_report_directory (Optional[str], optional): Default None. If str, the output directory where xml reports for each archive will be written.
761            content_management_policy (Union[None, str, bytes, bytearray, io.BytesIO, glasswall.content_management.policies.ArchiveManager], optional): The content management policy to apply.
762            raise_unsupported (bool, optional): Default True. Raise exceptions when Glasswall encounters an error. Fail silently if False.
763
764        Returns:
765            exported_archives_dict (dict): A dictionary of file paths relative to input_directory, and glasswall.GwReturnObj with attributes: "status" (int), "output_file" (bytes), "output_report" (bytes)
766        """
767        exported_archives_dict = {}
768        # Call export_archive on each file in input_directory to output_directory
769        for input_file in utils.list_file_paths(input_directory):
770            relative_path = os.path.relpath(input_file, input_directory)
771            # Construct paths for output file and output report
772            output_file = None if output_directory is None else os.path.join(os.path.abspath(output_directory), relative_path)
773            output_report = None if output_report_directory is None else os.path.join(os.path.abspath(output_report_directory), relative_path + ".xml")
774
775            result = self.export_archive(
776                input_file=input_file,
777                output_file=output_file,
778                output_report=output_report,
779                content_management_policy=content_management_policy,
780                raise_unsupported=raise_unsupported,
781            )
782
783            exported_archives_dict[relative_path] = result
784
785        return exported_archives_dict
786
787    def import_archive(self, input_file: Union[str, bytes, bytearray, io.BytesIO], output_file: Optional[str] = None, output_report: Optional[str] = None, content_management_policy: Union[None, str, bytes, bytearray, io.BytesIO, glasswall.content_management.policies.ArchiveManager] = None, include_analysis_report: Optional[bool] = False, raise_unsupported: Optional[bool] = True):
788        """ Imports an archive using the Glasswall engine.
789
790        Args:
791            input_file (Union[str, bytes, bytearray, io.BytesIO]): The archive file path or bytes.
792            output_file (Optional[str], optional): Default None. If str, write the archive to the output_file path.
793            output_report (Optional[str], optional): Default None. If str, write the analysis report to the output_report path.
794            content_management_policy (Union[None, str, bytes, bytearray, io.BytesIO, glasswall.content_management.policies.ArchiveManager], optional): The content management policy to apply.
795            include_analysis_report (Optional[bool], optional): Default False. If True, write the analysis report into the imported archive.
796            raise_unsupported (bool, optional): Default True. Raise exceptions when Glasswall encounters an error. Fail silently if False.
797
798        Returns:
799            gw_return_object (glasswall.GwReturnObj): An instance of class glasswall.GwReturnObj containing attributes including: "status" (int), "output_file" (bytes), "output_report" (bytes)
800        """
801        # Validate arg types
802        if not isinstance(input_file, (str, bytes, bytearray, io.BytesIO)):
803            raise TypeError(input_file)
804        if not isinstance(output_file, (type(None), str)):
805            raise TypeError(output_file)
806        if not isinstance(output_report, (type(None), str)):
807            raise TypeError(output_report)
808        if not isinstance(content_management_policy, (type(None), str, bytes, bytearray, io.BytesIO, glasswall.content_management.policies.policy.Policy)):
809            raise TypeError(content_management_policy)
810
811        # Convert string path arguments to absolute paths
812        if isinstance(input_file, str):
813            input_file = os.path.abspath(input_file)
814        # Convert string path arguments to absolute paths
815        if isinstance(output_file, str):
816            output_file = os.path.abspath(output_file)
817        if isinstance(output_report, str):
818            output_report = os.path.abspath(output_report)
819
820        # Convert inputs to bytes
821        if isinstance(input_file, str):
822            if not os.path.isfile(input_file):
823                raise FileNotFoundError(input_file)
824            with open(input_file, "rb") as f:
825                input_file_bytes = f.read()
826        elif isinstance(input_file, (bytes, bytearray, io.BytesIO)):
827            input_file_bytes = utils.as_bytes(input_file)
828
829        if isinstance(content_management_policy, str) and os.path.isfile(content_management_policy):
830            with open(content_management_policy, "rb") as f:
831                content_management_policy = f.read()
832        elif isinstance(content_management_policy, type(None)):
833            # Load default
834            content_management_policy = glasswall.content_management.policies.ArchiveManager(default="sanitise", default_archive_manager="process")
835        content_management_policy = utils.validate_xml(content_management_policy)
836
837        # API function declaration
838        self.library.GwFileImportArchive.argtypes = [
839            ct.c_void_p,  # void *inputBuffer
840            ct.c_size_t,  # size_t inputBufferLength
841            ct.POINTER(ct.c_void_p),  # void **outputFileBuffer
842            ct.POINTER(ct.c_size_t),  # size_t *outputFileBufferLength
843            ct.POINTER(ct.c_void_p),  # void **outputReportBuffer
844            ct.POINTER(ct.c_size_t),  # size_t *outputReportBufferLength
845            ct.c_char_p,  # const char *xmlConfigString
846            ct.c_int  # int includeAnalysisReports
847        ]
848
849        # Variable initialisation
850        gw_return_object = glasswall.GwReturnObj()
851        gw_return_object.input_buffer = ct.create_string_buffer(input_file_bytes)
852        gw_return_object.input_buffer_length = ct.c_size_t(len(input_file_bytes))
853        gw_return_object.output_buffer = ct.c_void_p()
854        gw_return_object.output_buffer_length = ct.c_size_t()
855        gw_return_object.output_report_buffer = ct.c_void_p()
856        gw_return_object.output_report_buffer_length = ct.c_size_t()
857        gw_return_object.content_management_policy = ct.c_char_p(content_management_policy.encode())
858        gw_return_object.include_analysis_report = ct.c_int(int(include_analysis_report))
859
860        with utils.CwdHandler(new_cwd=self.library_path):
861            # API call
862            gw_return_object.status = self.library.GwFileImportArchive(
863                gw_return_object.input_buffer,
864                gw_return_object.input_buffer_length,
865                ct.byref(gw_return_object.output_buffer),
866                ct.byref(gw_return_object.output_buffer_length),
867                ct.byref(gw_return_object.output_report_buffer),
868                ct.byref(gw_return_object.output_report_buffer_length),
869                gw_return_object.content_management_policy,
870                gw_return_object.include_analysis_report
871            )
872
873        if gw_return_object.output_buffer and gw_return_object.output_buffer_length:
874            gw_return_object.output_file = utils.buffer_to_bytes(
875                gw_return_object.output_buffer,
876                gw_return_object.output_buffer_length
877            )
878        if gw_return_object.output_report_buffer and gw_return_object.output_report_buffer_length:
879            gw_return_object.output_report = utils.buffer_to_bytes(
880                gw_return_object.output_report_buffer,
881                gw_return_object.output_report_buffer_length
882            )
883
884        # Write output file
885        if hasattr(gw_return_object, "output_file"):
886            if isinstance(output_file, str):
887                os.makedirs(os.path.dirname(output_file), exist_ok=True)
888                with open(output_file, "wb") as f:
889                    f.write(gw_return_object.output_file)
890
891        # Write output report
892        if hasattr(gw_return_object, "output_report"):
893            if isinstance(output_report, str):
894                os.makedirs(os.path.dirname(output_report), exist_ok=True)
895                with open(output_report, "wb") as f:
896                    f.write(gw_return_object.output_report)
897
898        input_file_repr = f"{type(input_file)} length {len(input_file)}" if isinstance(input_file, (bytes, bytearray,)) else input_file.__sizeof__() if isinstance(input_file, io.BytesIO) else input_file
899        if gw_return_object.status not in successes.success_codes:
900            log.error(f"\n\tinput_file: {input_file_repr}\n\toutput_file: {output_file}\n\tstatus: {gw_return_object.status}")
901            if raise_unsupported:
902                raise errors.error_codes.get(gw_return_object.status, errors.UnknownErrorCode)(gw_return_object.status)
903        else:
904            log.debug(f"\n\tinput_file: {input_file_repr}\n\toutput_file: {output_file}\n\tstatus: {gw_return_object.status}")
905
906        self.release()
907
908        return gw_return_object
909
910    def import_directory(self, input_directory: str, output_directory: Optional[str], output_report_directory: Optional[str] = None, content_management_policy: Union[None, str, bytes, bytearray, io.BytesIO, glasswall.content_management.policies.ArchiveManager] = None, include_analysis_report: Optional[bool] = False, raise_unsupported: bool = True):
911        """ Calls import_archive on each file in input_directory. The imported archives are written to output_directory maintaining the same directory structure as input_directory.
912
913        Args:
914            input_directory (str): The input directory containing archives to import.
915            output_directory (Optional[str], optional): Default None. If str, the output directory where the archives will be written.
916            output_report_directory (Optional[str], optional): Default None. If str, the output directory where xml reports for each archive will be written.
917            content_management_policy (Union[None, str, bytes, bytearray, io.BytesIO, glasswall.content_management.policies.ArchiveManager], optional): The content management policy to apply.
918            include_analysis_report (Optional[bool], optional): Default False. If True, write the analysis report into the imported archive.
919            raise_unsupported (bool, optional): Default True. Raise exceptions when Glasswall encounters an error. Fail silently if False.
920
921        Returns:
922            imported_archives_dict (dict): A dictionary of file paths relative to input_directory, and glasswall.GwReturnObj with attributes: "status" (int), "output_file" (bytes), "output_report" (bytes)
923        """
924        imported_archives_dict = {}
925        # Call import_archive on each file in input_directory to output_directory
926        for input_file in utils.list_file_paths(input_directory):
927            relative_path = os.path.relpath(input_file, input_directory)
928            # Construct paths for output file and output report
929            output_file = None if output_directory is None else os.path.join(os.path.abspath(output_directory), relative_path)
930            output_report = None if output_report_directory is None else os.path.join(os.path.abspath(output_report_directory), relative_path + ".xml")
931
932            result = self.import_archive(
933                input_file=input_file,
934                output_file=output_file,
935                output_report=output_report,
936                content_management_policy=content_management_policy,
937                include_analysis_report=include_analysis_report,
938                raise_unsupported=raise_unsupported,
939            )
940
941            imported_archives_dict[relative_path] = result
942
943        return imported_archives_dict