D7net
Home
Console
Upload
information
Create File
Create Folder
About
Tools
:
/
proc
/
thread-self
/
root
/
opt
/
imunify360
/
venv
/
lib
/
python3.11
/
site-packages
/
defence360agent
/
Filename :
migrate.py
back
Copy
#!/opt/imunify360/venv/bin/python3 """This module import peewee_migrate and apply migrations, for Imunify-AV it's entrypoint for service""" import os import sys from logging import getLogger from peewee_migrate import migrator import defence360agent.internals.logger from defence360agent.application import app from defence360agent.application.settings import configure from defence360agent.contracts.config import Core from defence360agent.contracts.config import Model from defence360agent.contracts.config import bool_from_envvar from defence360agent.router import Router from defence360agent.subsys import systemd_notifier from defence360agent.model.instance import db from defence360agent.model import tls_check logger = getLogger(__name__) GO_SERVICE_NAME = "/usr/bin/imunify-resident" def is_go_supported(start_pkg): """Check if go service is supported""" # not supported in antivirus mode if start_pkg != "im360.run_resident": return False return True def apply_migrations(migrations_dirs, attached_dbs=tuple()): """Apply migrations: restructure db, config files, etc.""" logger.info("Applying database migrations...") systemd_notifier.notify(systemd_notifier.AgentState.MIGRATING) # prepare database to operate in WAL journal_mode and run migrations tls_check.reset() db.init(Model.PATH) for db_path, schema_name in attached_dbs: db.execute_sql(f"ATTACH '{db_path}' AS {schema_name}") try: with db.atomic("EXCLUSIVE"): router = Router( db, migrations_dirs=migrations_dirs, logger=logger, ) # HACK: Migrator uses global unconfigurable LOGGER, # overrride it, to use our logging settings migrator.LOGGER = logger router.run() finally: # close connection immediately since later this process # will be replaced by execv db.close() def run(*, start_pkg="defence360agent", configure=configure): """Entry point for Imunify-AV service. Apply migrations, and then replace process with {start_pkg}.run module.""" os.umask(Core.FILE_UMASK) configure() defence360agent.internals.logger.reconfigure() systemd_notifier.notify(systemd_notifier.AgentState.READY) apply_migrations(app.MIGRATIONS_DIRS, app.MIGRATIONS_ATTACHED_DBS) logger.info("Starting main process...") systemd_notifier.notify(systemd_notifier.AgentState.STARTING) use_go = bool_from_envvar( "I360_USE_GO", default=is_go_supported(start_pkg) ) if use_go: Core.GO_FLAG_FILE.touch(exist_ok=True) logger.info("Run imunify-resident service") os.execv( GO_SERVICE_NAME, [ GO_SERVICE_NAME, ] + sys.argv[1:], ) else: Core.GO_FLAG_FILE.unlink(missing_ok=True) os.execv( sys.executable, [sys.executable, "-m", "{}".format(start_pkg)] + sys.argv[1:], ) if __name__ == "__main__": run()