1#@!#!123s

https 1#@!#!123s
: / proc / thread-self / root / proc / thread-self / root / usr / share / imunify360 / scripts /
Filename : modsec_scan_real.py
back
#!/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()