ees_network_drive/files.py

#
# Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
# or more contributor license agreements. Licensed under the Elastic License 2.0;
# you may not use this file except in compliance with the Elastic License 2.0.
#
"""Module responsible for fetching the files from the Network Drives and returning a document with
    all file details in json format.
"""
import os
from pathlib import Path

import errno
import tempfile
import time

from dateutil.parser import parse
from tika.tika import TikaException

from . import adapter, constant
from .utils import extract, fetch_users_from_csv_file, hash_id

ACCESS_ALLOWED_TYPE = 0
ACCESS_DENIED_TYPE = 1
ACCESS_MASK_DENIED_WRITE_PERMISSION = 278
ACCESS_MASK_ALLOWED_WRITE_PERMISSION = 1048854
STATUS_NO_SUCH_FILE = 3221225487
STATUS_NO_SUCH_DEVICE = 3221225486
STATUS_OBJECT_NAME_NOT_FOUND = 3221225524
STATUS_OBJECT_PATH_NOT_FOUND = 3221225530


class Files:
    """This class fetches objects from Network Drives
    """

    def __init__(self, logger, config, client):
        self.logger = logger
        self.user_mapping = config.get_value("network_drive_enterprise_search.user_mapping")
        self.drive_path = config.get_value("network_drive.path")
        self.server_ip = config.get_value("network_drive.server_ip")
        self.enable_document_permission = config.get_value("enable_document_permission")
        self.network_drives_client = client

    def is_file_present_on_network_drive(self, smb_connection, drive_name, folder_path,
                                         file_structure, ids_list, visited_folders, deleted_folders):
        """Checks that folder/file present in Network Drives or not
            :param smb_connection: connection object
            :param drive_name: service name of the Network Drives
            :param folder_path: the relative path of the folder
            :param file_structure: dictionary containing folder and list of files inside the folder
            :param ids_list: list of id's of deleted files
            :param visited_folders: list of visited path of folders
            :param deleted_folders: list of deleted path of folders
            Returns:
                folder_deleted: boolean value indicating folder is deleted or not
        """
        folder_deleted = False
        try:
            drive_path = Path(self.drive_path)
            available_files = smb_connection.listPath(drive_path.parts[0], folder_path)
            for file in available_files:
                if file_structure[folder_path].get(file.filename):
                    file_structure[folder_path].pop(file.filename)
            ids_list.extend(list(file_structure[folder_path].values()))
            visited_folders.append(folder_path)
        except Exception as exception:
            status = exception.smb_messages[-1].status
            if status in [STATUS_NO_SUCH_FILE, STATUS_NO_SUCH_DEVICE, STATUS_OBJECT_NAME_NOT_FOUND]:
                for folder in file_structure.keys():
                    if folder_path in folder:
                        deleted_folders.append(folder)
                        self.logger.info(f"{folder} entire folder is deleted.")
                deleted_folders.append(folder_path)
                return True
            elif status == STATUS_OBJECT_PATH_NOT_FOUND:
                folder_path, _ = os.path.split(folder_path)
                folder_deleted = self.is_file_present_on_network_drive(smb_connection, drive_name, folder_path,
                                                                       file_structure,
                                                                       ids_list, visited_folders, deleted_folders)
            else:
                self.logger.exception(f"Error while retrieving files from drive {drive_name}.Error: {exception}")
        return folder_deleted

    def recursive_fetch(self, smb_connection, service_name, path, store):
        """This method is used to recursively fetch folder paths from Network Drives.
            :param smb_connection: SMB connection object
            :param service_name: name of the drive
            :param path: Path of the Network Drives
            :param store: temporary storage for fetched ids from Drive
            :returns: list of all the folder paths in network drives
        """
        try:
            file_list = smb_connection.listPath(service_name, rf'{path}', search=16)
        except Exception as exception:
            self.logger.exception(f"Unknown error while fetching files {exception}")
            return store
        for file in file_list:
            if file.filename not in ['.', '..']:
                file_name = file.filename
                self.recursive_fetch(smb_connection, service_name, os.path.join(path, file_name), store)
        store.append(path)
        return store

    def extract_files(self, smb_connection, service_name, path, time_range, indexing_rules):
        """
            :param smb_connection: SMB connection object
            :param service_name: name of the drive
            :param path: Path of the Network Drives
            :param time_range: Start and End Time
            :param indexing_rules: object of indexing_rules
            :returns: dictionary of ids and file details for the files fetched
        """
        storage = {}
        try:
            file_list = smb_connection.listPath(service_name, rf'{path}')
        except Exception as exception:
            self.logger.exception(f"Unknown error while extracting files from folder {path}.Error {exception}")
            return storage
        for file in file_list:
            if not file.isDirectory:
                file_name = file.filename
                updated_at = \
                    time.strftime(constant.RFC_3339_DATETIME_FORMAT, time.gmtime(file.last_attr_change_time))
                created_at = \
                    time.strftime(constant.RFC_3339_DATETIME_FORMAT, time.gmtime(file.create_time))
                file_path = os.path.join(path, file_name)
                file_details = {
                    'updated_at': updated_at,
                    'file_type': os.path.splitext(file_name)[1],
                    'file_size': file.file_size,
                    'created_at': created_at,
                    'file_name': file_name,
                    'file_path': file_path,
                    'web_path': f"file://{self.server_ip}/{service_name}/{file_path}"
                }
                is_indexable = indexing_rules.should_index(file_details)
                if is_indexable \
                    and parse(time_range.get('start_time')) < parse(updated_at) \
                        and parse(updated_at) <= parse(time_range.get('end_time')):
                    file_id = file.file_id if file.file_id else hash_id(file_name, path)
                    storage.update({file_id: file_details})
        return storage

    def retrieve_permission(self, smb_connection, service_name, file_path):
        """This method is used to retrieve permission from Network Drives.
            :param smb_connection: SMB connection object
            :param service_name: name of the drive
            :param file_path: Path of the Network Drives
            :returns: hash of allow and deny permissions lists
        """
        try:
            security_info = smb_connection.getSecurity(service_name, rf'{file_path}')
        except Exception as exception:
            self.logger.exception(f"Unknown error while fetching permission details for file {file_path}.\
            Error {exception}")
            return [], []
        allow_users = []
        deny_users = []
        if security_info.dacl:
            aces = security_info.dacl.aces
            for ace in (aces or []):
                sid = str(ace.sid)
                if ace.type == ACCESS_ALLOWED_TYPE or ace.mask == ACCESS_MASK_DENIED_WRITE_PERMISSION:
                    allow_users.append(sid)
                if (ace.type == ACCESS_DENIED_TYPE and ace.mask != ACCESS_MASK_DENIED_WRITE_PERMISSION) or \
                        ace.mask == ACCESS_MASK_ALLOWED_WRITE_PERMISSION:
                    deny_users.append(sid)
                if not fetch_users_from_csv_file(self.user_mapping, self.logger).get(sid):
                    self.logger.warning(f"No mapping found for sid:{sid} in csv file. \
                        Please add the sid->user mapping for the {sid} and rerun the \
                        permission_sync_command to sync the user mappings.")
        return {'allow': allow_users, 'deny': deny_users}

    def fetch_files(self, service_name, path_list, time_range, indexing_rules):
        """This method is used to fetch and index files to Workplace Search
            :param service_name: name of the drive
            :param path_list: list of folder paths inside the network drives
            :param time_range: Start and End Time
            :param indexing_rules: object of indexing_rules
        """
        schema = adapter.FILES
        documents = []
        smb_connection = self.network_drives_client.connect()
        if smb_connection:
            for folder_path in path_list:
                storage = self.extract_files(smb_connection, service_name, folder_path, time_range, indexing_rules)
                for file_id, file_details in storage.items():
                    doc = {}
                    for field, file_field in schema.items():
                        doc[field] = file_details.get(file_field)
                    doc.update({'body': {}, 'id': str(file_id)})
                    if self.enable_document_permission:
                        permissions = self.retrieve_permission(
                            smb_connection, service_name, file_details.get("file_path"))
                        doc['_allow_permissions'] = permissions['allow']
                        doc['_deny_permissions'] = permissions['deny']
                    doc['body'] = self.fetch_file_content(service_name, file_details, smb_connection)
                    documents.append(doc)
            smb_connection.close()
        else:
            raise ConnectionError("Unknown error while connecting to network drives")
        return documents

    def fetch_file_content(self, service_name, file_details, smb_connection):
        """This method is used to fetch content from Network Drives file
        :param service_name: name of the drive
        :param file_details: dictionary containing file details
        :param smb_connection: connection object
        """
        file_obj = tempfile.NamedTemporaryFile()
        try:
            smb_connection.retrieveFile(service_name, file_details.get('file_path'), file_obj)
            file_obj.seek(0)
            try:
                extracted_content = extract(file_obj.read())
                file_obj.close()
                return extracted_content
            except TikaException as exception:
                self.logger.exception(
                    f"Error while extracting contents from file {file_details.get('file_name')} via Tika Parser. \
                        Error {exception}")
        except Exception as exception:
            if isinstance(exception, OSError) and exception.errno == errno.ENOSPC:
                self.logger.exception(
                    f"We reached the memory limit for extracting the file: {file_details.get('file_name')}. \
                        Skipping the contents of this file. Error: {exception}"
                )
            else:
                self.logger.exception(
                    f"Cannot read the contents of the file {file_details.get('file_name')} . Error {exception}")
        file_obj.close()