#!/opt/imunify360/venv/bin/python3 """Check whether a file is malicious. Usage: modsec-on-upload <file> It is expected to be called on a file upload by ModSecurity. It submits the file as an upload job to aibolit and waits for response (total timeout should be around aibolit_job.UPLOAD_TIMEOUT seconds). """ import signal import sys from im360 import aibolit_job __all__ = () def report_file_is_ok(): print(1, "OK", flush=True) def report_malicious_file(): print(0, "Attempt to upload malware", flush=True) def run(): # get parameters try: file_to_scan = sys.argv[1] except IndexError: print(__doc__, file=sys.stderr) # print usage sys.exit(2) resident_dir_path = aibolit_job.RESIDENT_DIR # to include the import time, we could read the start time of the # process https://gist.github.com/westhood/1073585 remaining_time = aibolit_job.create_remaining_time_func( aibolit_job.UPLOAD_TIMEOUT ) # signals we'll be waiting for from aibolit sigset = {signal.SIGUSR1, signal.SIGUSR2} # block the signal in all threads signal.pthread_sigmask(signal.SIG_BLOCK, sigset) # submit the uploaded file for scanning # create PID.upload_job in the resident dir aibolit_job.create_upload_job( files=[file_to_scan], resident_dir_path=resident_dir_path, timeout=remaining_time(), ) # notify aibolit about the new job aibolit_job.notify_aibolit_start_it_if_necessary( timeout=remaining_time() ) # wait for response while True: # use sigtimedwait() instead of signal() to get the uid # note: ignore a possible race on retry inside sigtimedwait() on # receiving a signal (see sigtimedwait()'s Python docs) si = signal.sigtimedwait(sigset, remaining_time()) if si is None: raise TimeoutError if si.si_uid == 0: # the signal is from root if si.si_signo == signal.SIGUSR1: return report_malicious_file() elif si.si_signo == signal.SIGUSR2: return report_file_is_ok() else: assert 0, "shouldn't happen" # pragma: no cover def main(): try: run() except: # noqa: assume it is # better to report a malicious file as ok then to reject a good file report_file_is_ok() raise if __name__ == "__main__": main()