X7ROOT File Manager
Current Path:
/opt/imunify360/venv/lib/python3.11/site-packages/imav/plugins
opt
/
imunify360
/
venv
/
lib
/
python3.11
/
site-packages
/
imav
/
plugins
/
π
..
π
__init__.py
(0 B)
π
__pycache__
π
check_license.py
(9.47 KB)
π
conflicts.py
(3.18 KB)
π
event_hook_executor.py
(3.72 KB)
π
event_hooks.py
(3.47 KB)
π
generic_sensor.py
(6.11 KB)
π
im360_register.py
(2.8 KB)
π
imunify_patch_id.py
(2.12 KB)
π
inotify.py
(1.75 KB)
π
malware_filters.py
(3.68 KB)
π
mr_proper.py
(2.67 KB)
π
plesk_notifications.py
(3.82 KB)
π
post_action.py
(1.72 KB)
π
restore_from_backup.py
(3.37 KB)
π
server_pull.py
(2.56 KB)
π
service_manager.py
(3.93 KB)
π
wordpress.py
(11.43 KB)
Editing: mr_proper.py
""" This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version. This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.Β See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with this program.Β If not, see <https://www.gnu.org/licenses/>. CopyrightΒ Β© 2019 Cloud Linux Software Inc. This software is also available under ImunifyAV commercial license, see <https://www.imunify360.com/legal/eula> """ from logging import getLogger from defence360agent.contracts.plugins import ( expect, MessageSource, MessageSink, ) from defence360agent.contracts.messages import MessageType from defence360agent.utils import recurring_check from defence360agent.utils.check_lock import check_lock from imav.malwarelib.scan.cleaners.detached_dir import ( MalwareDetachedDirCleaner, ) from imav.malwarelib.scan.cleaners.outdated_scans import ( OutdatedHistoryCleaner, OutdatedHitsCleaner, OutdatedScansCleaner, ) from imav.malwarelib.scan.cleaners.uncompleted_scans import ( UncompletedScansCleaner, ) logger = getLogger(__name__) class MrProper(MessageSource, MessageSink): CLEANERS = [ MalwareDetachedDirCleaner, OutdatedHistoryCleaner, OutdatedScansCleaner, OutdatedHitsCleaner, ] def __init__(self): self._loop = None self._sink = None self._tasks = [] async def create_source(self, loop, sink): self._loop = loop self._sink = sink for cleaner in self.CLEANERS: self._tasks.append( self._loop.create_task( recurring_check( check_lock, check_period_first=True, check_lock_period=cleaner.PERIOD, lock_file=cleaner.LOCK_FILE, )(cleaner.cleanup)() ) ) async def create_sink(self, loop): pass @expect(MessageType.CheckDetachedScans) async def check_detached_scans(self, _): await UncompletedScansCleaner.cleanup() logger.info("Detached scans checked") async def shutdown(self): for task in self._tasks: task.cancel() for task in self._tasks: # no need to suppress CancelledError because # recurring_check() catches it already await task
Upload File
Create Folder