BaseTools: Replace BSD License with BSD+Patent License

[mirror_edk2.git] / BaseTools / Source / Python / UPT / Core / FileHook.py

## @file\r
# This file hooks file and directory creation and removal\r
#\r
# Copyright (c) 2014 - 2018, Intel Corporation. All rights reserved.<BR>\r
#\r
# SPDX-License-Identifier: BSD-2-Clause-Patent\r
#\r
\r
'''\r
File hook\r
'''\r
\r
import os\r
import stat\r
import time\r
import zipfile\r
from time import sleep\r
from Library import GlobalData\r
\r
__built_in_remove__ = os.remove\r
__built_in_mkdir__  = os.mkdir\r
__built_in_rmdir__  = os.rmdir\r
__built_in_chmod__  = os.chmod\r
__built_in_open__   = open\r
\r
_RMFILE      = 0\r
_MKFILE      = 1\r
_RMDIR       = 2\r
_MKDIR       = 3\r
_CHMOD       = 4\r
\r
gBACKUPFILE = 'file.backup'\r
gEXCEPTION_LIST = ['Conf'+os.sep+'DistributionPackageDatabase.db', '.tmp', gBACKUPFILE]\r
\r
class _PathInfo:\r
    def __init__(self, action, path, mode=-1):\r
        self.action = action\r
        self.path = path\r
        self.mode = mode\r
\r
class RecoverMgr:\r
    def __init__(self, workspace):\r
        self.rlist = []\r
        self.zip = None\r
        self.workspace = os.path.normpath(workspace)\r
        self.backupfile = gBACKUPFILE\r
        self.zipfile = os.path.join(self.workspace, gBACKUPFILE)\r
\r
    def _createzip(self):\r
        if self.zip:\r
            return\r
        self.zip = zipfile.ZipFile(self.zipfile, 'w', zipfile.ZIP_DEFLATED)\r
\r
    def _save(self, tmp, path):\r
        if not self._tryhook(path):\r
            return\r
        self.rlist.append(_PathInfo(tmp, path))\r
\r
    def bkrmfile(self, path):\r
        arc = self._tryhook(path)\r
        if arc and os.path.isfile(path):\r
            self._createzip()\r
            self.zip.write(path, arc.encode('utf_8'))\r
            sta = os.stat(path)\r
            oldmode = stat.S_IMODE(sta.st_mode)\r
            self.rlist.append(_PathInfo(_CHMOD, path, oldmode))\r
            self.rlist.append(_PathInfo(_RMFILE, path))\r
        __built_in_remove__(path)\r
\r
    def bkmkfile(self, path, mode, bufsize):\r
        if not os.path.exists(path):\r
            self._save(_MKFILE, path)\r
        return __built_in_open__(path, mode, bufsize)\r
\r
    def bkrmdir(self, path):\r
        if os.path.exists(path):\r
            sta = os.stat(path)\r
            oldmode = stat.S_IMODE(sta.st_mode)\r
            self.rlist.append(_PathInfo(_CHMOD, path, oldmode))\r
            self._save(_RMDIR, path)\r
        __built_in_rmdir__(path)\r
\r
    def bkmkdir(self, path, mode):\r
        if not os.path.exists(path):\r
            self._save(_MKDIR, path)\r
        __built_in_mkdir__(path, mode)\r
\r
    def bkchmod(self, path, mode):\r
        if self._tryhook(path) and os.path.exists(path):\r
            sta = os.stat(path)\r
            oldmode = stat.S_IMODE(sta.st_mode)\r
            self.rlist.append(_PathInfo(_CHMOD, path, oldmode))\r
        __built_in_chmod__(path, mode)\r
\r
    def rollback(self):\r
        if self.zip:\r
            self.zip.close()\r
            self.zip = None\r
        index = len(self.rlist) - 1\r
        while index >= 0:\r
            item = self.rlist[index]\r
            exist = os.path.exists(item.path)\r
            if item.action == _MKFILE and exist:\r
                #if not os.access(item.path, os.W_OK):\r
                #    os.chmod(item.path, S_IWUSR)\r
                __built_in_remove__(item.path)\r
            elif item.action == _RMFILE and not exist:\r
                if not self.zip:\r
                    self.zip = zipfile.ZipFile(self.zipfile, 'r', zipfile.ZIP_DEFLATED)\r
                arcname = os.path.normpath(item.path)\r
                arcname = arcname[len(self.workspace)+1:].encode('utf_8')\r
                if os.sep != "/" and os.sep in arcname:\r
                    arcname = arcname.replace(os.sep, '/')\r
                mtime = self.zip.getinfo(arcname).date_time\r
                content = self.zip.read(arcname)\r
                filep = __built_in_open__(item.path, "wb")\r
                filep.write(content)\r
                filep.close()\r
                intime = time.mktime(mtime + (0, 0, 0))\r
                os.utime(item.path, (intime, intime))\r
            elif item.action == _MKDIR and exist:\r
                while True:\r
                    try:\r
                        __built_in_rmdir__(item.path)\r
                        break\r
                    except IOError:\r
                        # Sleep a short time and try again\r
                        # The anti-virus software may delay the file removal in this directory\r
                        sleep(0.1)\r
            elif item.action == _RMDIR and not exist:\r
                __built_in_mkdir__(item.path)\r
            elif item.action == _CHMOD and exist:\r
                try:\r
                    __built_in_chmod__(item.path, item.mode)\r
                except EnvironmentError:\r
                    pass\r
            index -= 1\r
        self.commit()\r
\r
    def commit(self):\r
        if self.zip:\r
            self.zip.close()\r
            __built_in_remove__(self.zipfile)\r
\r
    # Check if path needs to be hooked\r
    def _tryhook(self, path):\r
        path = os.path.normpath(path)\r
        works = self.workspace if str(self.workspace).endswith(os.sep) else (self.workspace  + os.sep)\r
        if not path.startswith(works):\r
            return ''\r
        for exceptdir in gEXCEPTION_LIST:\r
            full = os.path.join(self.workspace, exceptdir)\r
            if full == path or path.startswith(full + os.sep) or os.path.split(full)[0] == path:\r
                return ''\r
        return path[len(self.workspace)+1:]\r
\r
def _hookrm(path):\r
    if GlobalData.gRECOVERMGR:\r
        GlobalData.gRECOVERMGR.bkrmfile(path)\r
    else:\r
        __built_in_remove__(path)\r
\r
def _hookmkdir(path, mode=0o777):\r
    if GlobalData.gRECOVERMGR:\r
        GlobalData.gRECOVERMGR.bkmkdir(path, mode)\r
    else:\r
        __built_in_mkdir__(path, mode)\r
\r
def _hookrmdir(path):\r
    if GlobalData.gRECOVERMGR:\r
        GlobalData.gRECOVERMGR.bkrmdir(path)\r
    else:\r
        __built_in_rmdir__(path)\r
\r
def _hookmkfile(path, mode='r', bufsize=-1):\r
    if GlobalData.gRECOVERMGR:\r
        return GlobalData.gRECOVERMGR.bkmkfile(path, mode, bufsize)\r
    return __built_in_open__(path, mode, bufsize)\r
\r
def _hookchmod(path, mode):\r
    if GlobalData.gRECOVERMGR:\r
        GlobalData.gRECOVERMGR.bkchmod(path, mode)\r
    else:\r
        __built_in_chmod__(path, mode)\r
\r
def SetRecoverMgr(mgr):\r
    GlobalData.gRECOVERMGR = mgr\r
\r
os.remove   = _hookrm\r
os.mkdir    = _hookmkdir\r
os.rmdir    = _hookrmdir\r
os.chmod    = _hookchmod\r
__FileHookOpen__    = _hookmkfile\r