glasswall.libraries.security_tagging.security_tagging

  1import ctypes as ct
  2import os
  3import sys
  4
  5from glasswall import utils
  6from glasswall.config.logging import log
  7from glasswall.libraries.library import Library
  8from glasswall.libraries.security_tagging import errors, successes
  9
 10
 11class SecurityTagging(Library):
 12    """ A high level Python wrapper for Glasswall Security Tagging. """
 13
 14    def __init__(self, library_path: str):
 15        super().__init__(library_path=library_path)
 16        self.library = self.load_library(os.path.abspath(library_path))
 17
 18        log.info(f"Loaded Glasswall {self.__class__.__name__} version {self.version()} from {self.library_path}")
 19
 20    def version(self):
 21        """ Returns the Glasswall library version.
 22
 23        Returns:
 24            version (str): The Glasswall library version.
 25        """
 26        # TODO security tagging currently has no version function
 27        return "NOT_IMPLEMENTED"
 28
 29    def tag_file(self, tags_path: str, input_file: str, output_file: str, raise_unsupported: bool = True):
 30        """ Tags the input_file with xml loaded from tags_path, writing to output_file.
 31
 32        Args:
 33            tags_path (str): The path to the .xml file containing tags to add.
 34            input_file (str): The path to the input file.
 35            output_file (str): The path to the output file where the tagged input_file will be written to.
 36            raise_unsupported (bool, optional): Default True. Raise exceptions when Glasswall encounters an error. Fail silently if False.
 37
 38        Returns:
 39            status (int): An integer indicating the file process status.
 40
 41        Raises:
 42            TypeError: If any arguments are of incorrect type.
 43            NotImplementedError: If raise_unsupported is True and the status of calling GWSecuTag_TagFile is not 1 (success)
 44        """
 45        # Validate arg types
 46        if not isinstance(tags_path, str):
 47            raise TypeError(tags_path)
 48        elif not os.path.isfile(tags_path):
 49            raise FileNotFoundError(tags_path)
 50
 51        if not isinstance(input_file, str):
 52            raise TypeError(input_file)
 53        elif not os.path.isfile(input_file):
 54            raise FileNotFoundError(input_file)
 55
 56        if not isinstance(output_file, str):
 57            raise TypeError(output_file)
 58
 59        if not isinstance(raise_unsupported, bool):
 60            raise TypeError(raise_unsupported)
 61
 62        # Convert paths to absolute paths
 63        tags_path = os.path.abspath(tags_path)
 64        input_file = os.path.abspath(input_file)
 65        output_file = os.path.abspath(output_file)
 66
 67        # Make output directory if it does not exist
 68        os.makedirs(os.path.dirname(output_file), exist_ok=True)
 69
 70        with utils.CwdHandler(self.library_path):
 71            # API function declaration
 72            self.library.GWSecuTag_TagFile.argtypes = [ct.c_char_p]
 73
 74            # Variable initialisation
 75            ct_tags_path = ct.c_char_p(tags_path.encode("utf-8"))
 76            ct_input_file = ct.c_char_p(input_file.encode("utf-8"))
 77            ct_output_file = ct.c_char_p(output_file.encode("utf-8"))
 78
 79            # API call
 80            status = self.library.GWSecuTag_TagFile(
 81                ct_input_file,
 82                ct_tags_path,
 83                ct_output_file,
 84            )
 85
 86            if status not in successes.success_codes:
 87                log.error(f"\n\tinput_file: {input_file}\n\toutput_file: {output_file}\n\tstatus: {status}\n\ttags_path: {tags_path}")
 88                if raise_unsupported:
 89                    raise errors.error_codes.get(status, errors.UnknownErrorCode)(status)
 90            else:
 91                log.debug(f"\n\tinput_file: {input_file}\n\toutput_file: {output_file}\n\tstatus: {status}\n\ttags_path: {tags_path}")
 92
 93            # TODO remove, temp fix - check if tags are retrievable, if not delete the file just created by glasswall
 94            # if status not in successes.success_codes: # status is currently incorrect in the library
 95            if not os.path.isfile(output_file):
 96                log.error(f"\n\toutput file does not exist: {output_file}")
 97                status = "OUTPUT_NOT_CREATED"
 98            elif os.path.isfile(output_file):
 99                with utils.TempFilePath() as temp_file:
100                    self.retrieve_tags(
101                        input_file=output_file,
102                        output_file=temp_file,
103                        raise_unsupported=False
104                    )
105                    try:
106                        dict_ = utils.xml_as_dict(temp_file)
107                    except ValueError:
108                        dict_ = {}
109                    if not dict_:
110                        os.remove(output_file)
111                        log.debug(f"\n\tunable to retrieve tags, deleted output file\n\toutput_file: {output_file}\n\t")
112                        status = "OUTPUT_NOT_RETRIEVABLE"
113
114            return status
115
116    def tag_directory(self, tags_path: str, input_directory: str, output_directory: str, raise_unsupported: bool = True):
117        """ Tags all files in input_directory with the xml loaded from tags_path, writing to output_directory and maintaining the same directory structure.
118
119        Args:
120            tags_path (str): The path to the .xml file containing tags to add.
121            input_directory (str): The path to the input directory.
122            output_directory (str): The path to the output directory where the tagged files will be written to.
123            raise_unsupported (bool, optional): Default True. Raise exceptions when Glasswall encounters an error. Fail silently if False.
124
125        Returns:
126            status (int): An integer indicating the file process status.
127
128        Raises:
129            TypeError: If any arguments are of incorrect type.
130            NotImplementedError: If raise_unsupported is True and the status of calling GWSecuTag_TagFile is not 1 (success)
131        """
132        # Validate arg types
133        if not isinstance(tags_path, str):
134            raise TypeError(tags_path)
135        elif not os.path.isfile(tags_path):
136            raise FileNotFoundError(tags_path)
137
138        if not isinstance(input_directory, str):
139            raise TypeError(input_directory)
140        elif not os.path.isdir(input_directory):
141            raise NotADirectoryError(input_directory)
142
143        if not isinstance(output_directory, str):
144            raise TypeError(output_directory)
145
146        if not isinstance(raise_unsupported, bool):
147            raise TypeError(raise_unsupported)
148
149        for relative_path in utils.list_file_paths(input_directory, absolute=False):
150            # construct absolute paths
151            input_file = os.path.abspath(os.path.join(input_directory, relative_path))
152            output_file = os.path.abspath(os.path.join(output_directory, relative_path))
153
154            # call tag_file on each file in input to output
155            self.tag_file(
156                tags_path=tags_path,
157                input_file=input_file,
158                output_file=output_file,
159                raise_unsupported=raise_unsupported,
160            )
161
162        utils.delete_empty_subdirectories(output_directory)
163
164    def retrieve_tags(self, input_file: str, output_file: str, raise_unsupported=True):
165        """ Retrieves the xml tags of the input_file and writes it to output_file.
166
167        Args:
168            input_file (str): The path to the input file.
169            output_file (str): The path to the output file where the xml tags will be written to.
170            raise_unsupported (bool, optional): Default True. Raise exceptions when Glasswall encounters an error. Fail silently if False.
171
172        Returns:
173            status (int): An integer indicating the file process status.
174
175        Raises:
176            TypeError: If any arguments are of incorrect type.
177            NotImplementedError: If raise_unsupported is True and the status of calling GWSecuTag_RetrieveTagFile is not 1 (success)
178        """
179        if not isinstance(input_file, str):
180            raise TypeError(input_file)
181        elif not os.path.isfile(input_file):
182            raise FileNotFoundError(input_file)
183
184        if not isinstance(output_file, str):
185            raise TypeError(output_file)
186
187        # Convert paths to absolute paths
188        input_file = os.path.abspath(input_file)
189        output_file = os.path.abspath(output_file)
190
191        # Make output directory if it does not exist
192        os.makedirs(os.path.dirname(output_file), exist_ok=True)
193
194        log.debug(f"Attempting {sys._getframe().f_code.co_name}:\n\tinput_file: {input_file}\n\toutput_file: {output_file}")
195
196        with utils.CwdHandler(self.library_path):
197            # API function declaration
198            self.library.GWSecuTag_RetrieveTagFile.argtypes = [ct.c_char_p]
199
200            # Variable initialisation
201            ct_input_file = ct.c_char_p(input_file.encode("utf-8"))
202            ct_output_file = ct.c_char_p(output_file.encode("utf-8"))
203
204            # API call
205            status = self.library.GWSecuTag_RetrieveTagFile(
206                ct_input_file,
207                ct_output_file,
208            )
209
210            if status not in successes.success_codes:
211                log.error(f"\n\tinput_file: {input_file}\n\toutput_file: {output_file}\n\tstatus: {status}")
212                if raise_unsupported:
213                    raise errors.error_codes.get(status, errors.UnknownErrorCode)(status)
214            else:
215                log.debug(f"\n\tinput_file: {input_file}\n\toutput_file: {output_file}\n\tstatus: {status}")
216
217            return status
218
219    def retrieve_tags_directory(self, input_directory: str, output_directory: str, raise_unsupported=True):
220        """ Retrieves all tags from files in input_directory, writing XML to output_directory and maintaining the same directory structure.
221
222        Args:
223            input_directory (str): The path to the input directory.
224            output_directory (str): The path to the output directory where the tagged files will be written to.
225            raise_unsupported (bool, optional): Default True. Raise exceptions when Glasswall encounters an error. Fail silently if False.
226
227        Returns:
228            status (int): The status of the function call.
229
230        Raises:
231            TypeError: If any arguments are of incorrect type.
232            NotImplementedError: If raise_unsupported is True and the status of calling GWSecuTag_RetrieveTagFile is not 1 (success)
233        """
234        # Validate arg types
235        if not isinstance(input_directory, str):
236            raise TypeError(input_directory)
237        elif not os.path.isdir(input_directory):
238            raise NotADirectoryError(input_directory)
239
240        if not isinstance(output_directory, str):
241            raise TypeError(output_directory)
242
243        if not isinstance(raise_unsupported, bool):
244            raise TypeError(raise_unsupported)
245
246        for relative_path in utils.list_file_paths(input_directory, absolute=False):
247            # construct absolute paths
248            input_file = os.path.abspath(os.path.join(input_directory, relative_path))
249            output_file = os.path.abspath(os.path.join(output_directory, relative_path + ".xml"))
250
251            # call retrieve_tags on each file in input to output
252            self.retrieve_tags(
253                input_file=input_file,
254                output_file=output_file,
255                raise_unsupported=raise_unsupported,
256            )
257
258        utils.delete_empty_subdirectories(output_directory)