clamav/unit_tests/clamd_test.py

# Copyright (C) 2020-2025 Cisco Systems, Inc. and/or its affiliates. All rights reserved.

"""
Run clamd (and clamdscan) tests.
"""

import os
from pathlib import Path
import platform
import socket
import subprocess
import shutil
import sys
import time
import unittest
from zipfile import ZIP_DEFLATED, ZipFile

import testcase


os_platform = platform.platform()
operating_system = os_platform.split('-')[0].lower()

def check_port_available(port_num: int) -> bool:
    '''
    Check if port # is available
    '''
    port_is_available = True # It's probably available...

    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    location = ('localhost', port_num)

    result_of_check = sock.connect_ex(location)
    if result_of_check == 0:
        port_is_available = False # Oh nevermind! Someone was listening!
    sock.close()

    return port_is_available

class TC(testcase.TestCase):
    @classmethod
    def setUpClass(cls):
        super(TC, cls).setUpClass()

        TC.testpaths = list(TC.path_build.glob('unit_tests/input/clamav_hdb_scanfiles/clam*')) # A list of Path()'s of each of our generated test files

        TC.clamd_pid = TC.path_tmp / 'clamd-test.pid'
        TC.clamd_socket =   'clamd-test.socket'             # <-- A relative path here and in check_clamd to avoid-
                                                            # test failures caused by (invalid) long socket filepaths.
                                                            # The max length for a socket file path is _really_ short.
        TC.clamd_port_num = 3319                            # <-- This is hard-coded into the `check_clamd` program
        TC.path_db = TC.path_tmp / 'database'
        TC.path_db.mkdir(parents=True)
        shutil.copy(
            str(TC.path_build / 'unit_tests' / 'input' / 'clamav.hdb'),
            str(TC.path_db),
        )
        shutil.copy(
            str(TC.path_source / 'unit_tests' / 'input' / 'other_sigs' / 'daily.pdb'),
            str(TC.path_db),
        )

        # Identify a TCP port we can use.
        # Presently disabled because check_clamd's port # is hardcoded.
        #found_open_port = False
        #for port_num in range(3310, 3410):
        #    if check_port_available(port_num) == True:
        #        found_open_port = True
        #        break
        #assert found_open_port == True

        # Prep a clamd.conf to use for most (if not all) of the tests.
        config = '''
            Foreground yes
            PidFile {pid}
            DatabaseDirectory {dbdir}
            LogFileMaxSize 0
            LogTime yes
            #Debug yes
            LogClean yes
            LogVerbose yes
            ExitOnOOM yes
            DetectPUA yes
            ScanPDF yes
            CommandReadTimeout 1
            MaxQueue 800
            MaxConnectionQueueLength 1024
            '''.format(pid=TC.clamd_pid, dbdir=TC.path_db)
        if operating_system == 'windows':
            # Only have TCP socket option for Windows.
            config += '''
                TCPSocket {socket}
                TCPAddr localhost
                '''.format(socket=TC.clamd_port_num)
        else:
            # Use LocalSocket for Posix, because that's what check_clamd expects.
            config += '''
                LocalSocket {localsocket}
                TCPSocket {tcpsocket}
                TCPAddr localhost
                '''.format(localsocket=TC.clamd_socket, tcpsocket=TC.clamd_port_num)

        TC.clamd_config = TC.path_tmp / 'clamd-test.conf'
        TC.clamd_config.write_text(config)

        # Check if fdpassing is supported.
        TC.has_fdpass_support = False
        with (TC.path_build / 'clamav-config.h').open('r') as clamav_config:
            if "#define HAVE_FD_PASSING 1" in clamav_config.read():
                TC.has_fdpass_support = True

    @classmethod
    def tearDownClass(cls):
        super(TC, cls).tearDownClass()

    def setUp(self):
        super(TC, self).setUp()
        self.proc = None

    def tearDown(self):
        super(TC, self).tearDown()

        # Kill clamd (if running)
        if self.proc != None:
            try:
                self.proc.terminate()
                self.proc.wait(timeout=120)
                self.proc.stdin.close()
            except OSError as exc:
                self.log.warning('Unexpected exception {}'.format(exc))
                pass  # ignore
            self.proc = None
        try:
            TC.clamd_pid.unlink()
        except Exception:
            pass # missing_ok=True is too for common use.
        try:
            TC.clamd_socket.unlink()
        except Exception:
            pass # missing_ok=True is too for common use.

        self.verify_valgrind_log()

    def start_clamd(self, use_valgrind=True, clamd_config=None):
        '''
        Start clamd
        '''
        if clamd_config == None:
            clamd_config = TC.clamd_config

        if use_valgrind:
            command = '{valgrind} {valgrind_args} {clamd} --config-file={clamd_config}'.format(
                valgrind=TC.valgrind, valgrind_args=TC.valgrind_args, clamd=TC.clamd, clamd_config=clamd_config
            )
        else:
            command = '{clamd} --config-file={clamd_config}'.format(
                clamd=TC.clamd, clamd_config=TC.clamd_config
            )
        self.log.info('Starting clamd: {}'.format(command))
        self.proc = subprocess.Popen(
            command.strip().split(' '),
            stdin=subprocess.PIPE,
            stdout=sys.stdout.buffer,
            stderr=sys.stdout.buffer,
        )

    def run_clamdscan(self,
                      scan_args,
                      expected_ec=0,
                      expected_out=[],
                      expected_err=[],
                      unexpected_out=[],
                      unexpected_err=[],
                      use_valgrind=False):
        '''
        Run clamdscan in each mode
        The first scan uses ping & wait to give clamd time to start.
        '''
        arg_variations = [
            '--ping 5 --wait',          # default (filepath) mode
            '--multiscan',              # multi mode
            '--stream',                 # stream mode
            '--stream --multiscan',     # fdstreampass multi mode
        ]
        if TC.has_fdpass_support:
            arg_variations += [
                '--fdpass',             # fdpass mode
                '--fdpass --multiscan', # fdpass multi mode
            ]

        for arg_variation in arg_variations:
            if use_valgrind:
                output = self.execute_command('{valgrind} {valgrind_args} {clamdscan} {arg_variation} {scan_args} -c {clamd_config}'.format(
                    valgrind=TC.valgrind, valgrind_args=TC.valgrind_args, clamdscan=TC.clamdscan, clamd_config=TC.clamd_config, arg_variation=arg_variation, scan_args=scan_args))
            else:
                output = self.execute_command('{clamdscan} {arg_variation} {scan_args} -c {clamd_config}'.format(
                    clamdscan=TC.clamdscan, clamd_config=TC.clamd_config, arg_variation=arg_variation, scan_args=scan_args))

            if expected_out != [] or unexpected_out != []:
                self.verify_output(output.out, expected=expected_out, unexpected=unexpected_out)
            if expected_err != [] or unexpected_err != []:
                self.verify_output(output.err, expected=expected_err, unexpected=unexpected_err)

            if use_valgrind:
                self.verify_valgrind_log()


    def run_clamdscan_file_only(self,
                                scan_args,
                                expected_ec=0,
                                expected_out=[],
                                expected_err=[],
                                unexpected_out=[],
                                unexpected_err=[]):
        '''
        Run clamdscan in filepath mode (and filepath multi mode)
        The first scan uses ping & wait to give clamd time to start.
        '''
        # default mode
        output = self.execute_command('{clamdscan} --ping 5 --wait -c {clamd_config} {scan_args}'.format(
            clamdscan=TC.clamdscan, clamd_config=TC.clamd_config, scan_args=scan_args))
        assert output.ec == expected_ec
        if expected_out != [] or unexpected_out != []:
            self.verify_output(output.out, expected=expected_out, unexpected=unexpected_out)
        if expected_err != [] or unexpected_err != []:
            self.verify_output(output.err, expected=expected_err, unexpected=unexpected_err)

        # multi mode
        output = self.execute_command('{clamdscan} -c {clamd_config} -m {scan_args}'.format(
            clamdscan=TC.clamdscan, clamd_config=TC.clamd_config, scan_args=scan_args))
        assert output.ec == expected_ec
        if expected_out != [] or unexpected_out != []:
            self.verify_output(output.out, expected=expected_out, unexpected=unexpected_out)
        if expected_err != [] or unexpected_err != []:
            self.verify_output(output.err, expected=expected_err, unexpected=unexpected_err)

    def run_clamdscan_fdpass_only(self,
                                  scan_args,
                                  expected_ec=0,
                                  expected_out=[],
                                  expected_err=[],
                                  unexpected_out=[],
                                  unexpected_err=[]):
        '''
        Run clamdscan fdpass mode only
        Use ping & wait to give clamd time to start.
        '''
        # fdpass
        output = self.execute_command('{clamdscan} --ping 5 --wait -c {clamd_config} --fdpass {scan_args}'.format(
            clamdscan=TC.clamdscan, clamd_config=TC.clamd_config, scan_args=scan_args))
        assert output.ec == expected_ec
        if expected_out != [] or unexpected_out != []:
            self.verify_output(output.out, expected=expected_out, unexpected=unexpected_out)
        if expected_err != [] or unexpected_err != []:
            self.verify_output(output.err, expected=expected_err, unexpected=unexpected_err)

    def test_clamd_00_version(self):
        '''
        verify that clamd -v returns the version
        '''
        self.step_name('clamd version test')

        command = '{valgrind} {valgrind_args} {clamd} --config-file={clamd_config} -V'.format(
            valgrind=TC.valgrind, valgrind_args=TC.valgrind_args, clamd=TC.clamd, clamd_config=TC.clamd_config)
        output = self.execute_command(command)

        assert output.ec == 0  # success

        expected_results = [
            'ClamAV {}'.format(TC.version),
        ]
        self.verify_output(output.out, expected=expected_results)

    def test_clamd_01_ping_pong(self):
        '''
        Verify that clamd responds to a PING command
        '''
        self.step_name('Testing clamd + clamdscan PING PONG feature')

        self.start_clamd()

        poll = self.proc.poll()
        assert poll == None  # subprocess is alive if poll() returns None

        output = self.execute_command('{clamdscan} -p 5 -c {clamd_config}'.format(
            clamdscan=TC.clamdscan, clamd_config=TC.clamd_config))

        assert output.ec == 0  # success
        self.verify_output(output.out, expected=['PONG'])

    def test_clamd_02_clamdscan_version(self):
        '''
        Verify that clamdscan --version returns the expected version #
        Explanation: clamdscan --version will query clamd for its version
          and print out clamd's version.  If it can't connect to clamd, it'll
          throw and error saying as much and then report its own version.

        In this test, we want to check clamd's version through clamdscan.
        '''
        self.step_name('Testing clamd + clamdscan version feature')

        self.start_clamd()

        poll = self.proc.poll()
        assert poll == None  # subprocess is alive if poll() returns None

        # First we'll ping-pong to make sure clamd is up
        # If clamd isn't up before the version test, clamdscan will return its
        # own version, which isn't really the point of the test.
        output = self.execute_command('{clamdscan} --ping 5 -c {clamd_config}'.format(
            clamdscan=TC.clamdscan, clamd_config=TC.clamd_config))
        assert output.ec == 0  # success
        self.verify_output(output.out, expected=['PONG'])

        # Ok now it's up, let's check clamd's version via clamdscan.
        output = self.execute_command('{clamdscan} --version -c {clamd_config}'.format(
            clamdscan=TC.clamdscan, clamd_config=TC.clamd_config))
        assert output.ec == 0  # success
        self.verify_output(output.out,
            expected=['ClamAV {}'.format(TC.version)], unexpected=['Could not connect to clamd'])

    def test_clamd_03_reload(self):
        '''
        In this test, it is not supposed to detect until we actually put the
        signature there and reload!
        '''
        self.step_name('Test scan before & after reload')

        self.start_clamd()

        poll = self.proc.poll()
        assert poll == None  # subprocess is alive if poll() returns None

        (TC.path_tmp / 'reload-testfile').write_bytes(b'ClamAV-RELOAD-Test')

        self.run_clamdscan('{}'.format(TC.path_tmp / 'reload-testfile'),
            expected_ec=0, expected_out=['reload-testfile: OK', 'Infected files: 0'])

        (TC.path_db / 'reload-test.ndb').write_text('ClamAV-RELOAD-TestFile:0:0:436c616d41562d52454c4f41442d54657374')

        output = self.execute_command('{clamdscan} --reload -c {clamd_config}'.format(
            clamdscan=TC.clamdscan, clamd_config=TC.clamd_config))
        assert output.ec == 0  # success

        time.sleep(2) # give clamd a moment to reload before trying again
                      # with multi-threaded reloading will clamd would happily
                      # re-scan with the old engine while it reloads.

        self.run_clamdscan('{}'.format(TC.path_tmp / 'reload-testfile'),
            expected_ec=1, expected_out=['ClamAV-RELOAD-TestFile.UNOFFICIAL FOUND', 'Infected files: 1'])

    def test_clamd_04_all_testfiles(self):
        '''
        Verify that clamd + clamdscan detect each of our <build>/unit_tests/input/clamav_hdb_scanfiles/clam* test files.
        '''
        self.step_name('Testing clamd + clamdscan scan of all `test` files')

        self.start_clamd()

        poll = self.proc.poll()
        assert poll == None  # subprocess is alive if poll() returns None

        testfiles = ' '.join([str(testpath) for testpath in TC.testpaths])
        expected_results = ['{}: ClamAV-Test-File.UNOFFICIAL FOUND'.format(testpath.name) for testpath in TC.testpaths]
        expected_results.append('Infected files: {}'.format(len(TC.testpaths)))

        self.run_clamdscan(testfiles,
            expected_ec=1, expected_out=expected_results)

    def test_clamd_05_check_clamd(self):
        '''
        Uses the check_clamd program to test clamd's socket API in various ways
        that aren't possible with clamdscan.
        '''
        self.step_name('Testing clamd + check_clamd')

        self.start_clamd()

        poll = self.proc.poll()
        assert poll == None  # subprocess is alive if poll() returns None

        # Let's first use the ping-pong test to make sure clamd is listening.
        output = self.execute_command('{clamdscan} -p 5 -c {clamd_config}'.format(
            clamdscan=TC.clamdscan, clamd_config=TC.clamd_config))
        assert output.ec == 0  # success
        self.verify_output(output.out, expected=['PONG'])

        # Ok now run check_clamd to have fun with clamd's API
        output = self.execute_command('{}'.format(TC.check_clamd))
        self.log.info('check_clamd stdout: \n{}'.format(output.out))
        self.log.info('check_clamd stderr: \n{}'.format(output.err))
        assert output.ec == 0  # success

        expected_results = [
            '100%', 'Failures: 0', 'Errors: 0'
        ]
        self.verify_output(output.out, expected=expected_results)

        # Let's do another ping-pong test to see if `check_clamd` killed clamd (Mu-ha-ha).
        output = self.execute_command('{clamdscan} -p 5 -c {clamd_config}'.format(
            clamdscan=TC.clamdscan, clamd_config=TC.clamd_config))
        assert output.ec == 0  # success
        self.verify_output(output.out, expected=['PONG'])

        time.sleep(5)

    def test_clamd_06_HeuristicScanPrecedence_off(self):
        '''
        Verify that HeuristicScanPrecedence off works as expected (default)
        In a later test, we'll add `HeuristicScanPrecedence yes` to the config
        and retest with it on.

        With it off, we expect the scan to complete and the "real" virus to alert
        rather than the heuristic.
        '''
        self.step_name('Testing clamd + clamdscan w/ HeuristicScanPrecedence no (default)')

        self.start_clamd()

        poll = self.proc.poll()
        assert poll == None  # subprocess is alive if poll() returns None

        self.run_clamdscan('{}'.format(TC.path_build / 'unit_tests' / 'input' / 'clamav_hdb_scanfiles' / 'clam.exe_and_mail.tar.gz'),
            expected_ec=1, expected_out=['ClamAV-Test-File'])

    def test_clamd_07_HeuristicScanPrecedence_on(self):
        '''
        Verify that HeuristicScanPrecedence on works as expected.

        With it on, we expect the scan to stop and raise an alert as soon as
        the phishing heuristic is detected.
        '''
        self.step_name('Testing clamd + clamdscan w/ HeuristicScanPrecedence yes')

        with TC.clamd_config.open('a') as config:
            config.write('''
                HeuristicScanPrecedence yes
                ''')

        self.start_clamd()

        poll = self.proc.poll()
        assert poll == None  # subprocess is alive if poll() returns None

        self.run_clamdscan('{}'.format(TC.path_build / 'unit_tests' / 'input' / 'clamav_hdb_scanfiles' / 'clam.exe_and_mail.tar.gz'),
            expected_ec=1, expected_out=['Heuristics.Phishing.Email.SpoofedDomain'])

    @unittest.skipIf(operating_system == 'windows', 'This test uses a shell script to test virus-action. TODO: add Windows support to this test.')
    def test_clamd_08_VirusEvent(self):
        '''
        Test that VirusEvent works
        '''
        self.step_name('Testing clamd + clamdscan w/ VirusEvent')

        with TC.clamd_config.open('a') as config:
            config.write('VirusEvent {} {} "Virus found: %v"\n'.format(
                TC.path_source / 'unit_tests' / 'input' / 'virusaction-test.sh',
                TC.path_tmp))

        self.start_clamd()

        poll = self.proc.poll()
        assert poll == None  # subprocess is alive if poll() returns None

        self.run_clamdscan_file_only('{}'.format(TC.path_build / 'unit_tests' / 'input' / 'clamav_hdb_scanfiles' / 'clam.exe'),
            expected_ec=1)#, expected_out=['Virus found: ClamAV-Test-File.UNOFFICIAL'])

        self.log.info('verifying log output from virusaction-test.sh: {}'.format(str(TC.path_tmp / 'test-clamd.log')))
        self.verify_log(str(TC.path_tmp / 'test-clamd.log'),
            expected=['Virus found: ClamAV-Test-File.UNOFFICIAL'],
            unexpected=['VirusEvent incorrect', 'VirusName incorrect'])

    def test_clamd_09_clamdscan_ExcludePath(self):
        '''
        Verify that ExcudePath works and does not cause other  on works as expected.
        We'll use valgrind on clamdscan instead of clamd for this one, if enabled
        as a regression for clamdscan memory leak fixes.

        With it on, we expect the scan to stop and raise an alert as soon as
        the phishing heuristic is detected.
        '''
        self.step_name('Testing clamd + clamdscan w/ ExcludePath')

        (TC.path_tmp / 'alpha').mkdir()
        (TC.path_tmp / 'beta').mkdir()
        (TC.path_tmp / 'charlie').mkdir()

        shutil.copy(str(TC.path_build / 'unit_tests' / 'input' / 'clamav_hdb_scanfiles' / 'clam.exe'), str(TC.path_tmp / 'alpha' / 'a_found'))     # This should be found (first)
        shutil.copy(str(TC.path_build / 'unit_tests' / 'input' / 'clamav_hdb_scanfiles' / 'clam.exe'), str(TC.path_tmp / 'beta' / 'b_excluded'))  # This one should be excluded
        shutil.copy(str(TC.path_build / 'unit_tests' / 'input' / 'clamav_hdb_scanfiles' / 'clam.exe'), str(TC.path_tmp / 'charlie' / 'c_found'))     # This one should still be found after excluding the previous

        with TC.clamd_config.open('a') as config:
            exclude_path = 'beta'

            config.write('''
                ExcludePath {}
                '''.format(exclude_path))

        self.start_clamd(use_valgrind=False)

        poll = self.proc.poll()
        assert poll == None  # subprocess is alive if poll() returns None

        expected_out = [
            'a_found: ClamAV-Test-File.UNOFFICIAL FOUND',
          # 'b_excluded: Excluded',  <-- Bug: this doesn't appear in recursive regular scans :-(, only fdpass and stream
            'c_found: ClamAV-Test-File.UNOFFICIAL FOUND',
        ]

        unexpected_out = [
            'a_found: Excluded',
            'b_excluded: ClamAV-Test-File.UNOFFICIAL FOUND',
            'c_found: Excluded',
        ]

        self.run_clamdscan('{}'.format(TC.path_tmp),
            expected_ec=1, expected_out=expected_out, unexpected_out=unexpected_out,
            use_valgrind=True)

    def test_clamd_10_allmatch_not_sticky(self):
        '''
        Verify that a scanning without allmatch does not use allmatch mode, after scanning with allmatch.
        This is a regression test for an issue where the allmatch scan option is sticky and any scans after an allmatch scan become an allmatch scan.
        '''
        self.step_name('Testing clamdscan --allmatch is not sticky')

        # Get a list of Path()'s of each of signature file
        test_path = TC.path_source / 'unit_tests' / 'input' / 'pe_allmatch'
        database_files = list(test_path.glob('alert-sigs/*'))

        test_exe = test_path / 'test.exe'

        # Copy them to the database directory before starting ClamD
        for db in database_files:
            shutil.copy(str(db), str(TC.path_db))

        #
        # Start ClamD
        #
        self.start_clamd()

        poll = self.proc.poll()
        assert poll == None  # subprocess is alive if poll() returns None

        # Try first without --allmatch
        output = self.execute_command('{clamdscan} -c {clamd_config} --wait --ping 10 {test_exe}'.format(
            clamdscan=TC.clamdscan, clamd_config=TC.clamd_config, test_exe=test_exe))
        assert output.ec == 1
        assert output.out.count('FOUND') == 1


        # Next, try WITH --allmatch
        output = self.execute_command('{clamdscan} -c {clamd_config} --allmatch {test_exe}'.format(
            clamdscan=TC.clamdscan, clamd_config=TC.clamd_config, test_exe=test_exe))
        assert output.ec == 1
        assert output.out.count('FOUND') > 1


        # Try again without --allmatch
        output = self.execute_command('{clamdscan} -c {clamd_config} {test_exe}'.format(
            clamdscan=TC.clamdscan, clamd_config=TC.clamd_config, test_exe=test_exe))
        assert output.ec == 1
        assert output.out.count('FOUND') == 1

    def test_clamd_11_alertexceedsmax_maxfilesize(self):
        '''
        Verify that exceeding maxfilesize with AlertExceedsMax reports an alert and not an error.
        We'll use some fine-tuned values to make sure we check both MaxFileSize (for a flat file) and MaxScanSize (for an archive).
        '''
        self.step_name('Testing clamd\'s AlertExceedsMax with MaxFileSize and MaxScanSize')

        # Make a "big" file to test to test max filesize.
        # We'll go with 501 bytes.
        big_file = TC.path_tmp / 'big_file'
        with big_file.open('wb') as test_file:
            test_file.write(b'\x00' * 501)

        # Make an even bigger archive to test max scansize,
        # This ends up being 492 bytes.
        big_zip = TC.path_tmp / 'big_zip'
        with ZipFile(str(big_zip), 'w', ZIP_DEFLATED) as zf:
            # Add a bunch of smaller files that won't exceed max filesize
            for i in range(0, 5):
                zf.writestr(f'file-{i}', b'\x00' * 50)

        # We'll use a config that sets:
        #   MaxFileSize 500
        #   MaxScanSize 500
        #   AlertExceedsMax yes
        config = '''
            Foreground yes
            PidFile {pid}
            DatabaseDirectory {dbdir}
            LogFileMaxSize 0
            LogTime yes
            #Debug yes
            LogClean yes
            LogVerbose yes
            ExitOnOOM yes
            DetectPUA yes
            ScanPDF yes
            CommandReadTimeout 1
            MaxQueue 800
            MaxConnectionQueueLength 1024
            MaxFileSize 500
            MaxScanSize 500
            AlertExceedsMax yes
            '''.format(pid=TC.clamd_pid, dbdir=TC.path_db)
        if operating_system == 'windows':
            # Only have TCP socket option for Windows.
            config += '''
                TCPSocket {socket}
                TCPAddr localhost
                '''.format(socket=TC.clamd_port_num)
        else:
            # Use LocalSocket for Posix, because that's what check_clamd expects.
            config += '''
                LocalSocket {localsocket}
                TCPSocket {tcpsocket}
                TCPAddr localhost
                '''.format(localsocket=TC.clamd_socket, tcpsocket=TC.clamd_port_num)

        clamd_config = TC.path_tmp / 'clamd-test.conf'
        clamd_config.write_text(config)

        # Copy database to database path
        shutil.copy(
            str(TC.path_build / 'unit_tests' / 'input' / 'clamav.hdb'),
            str(TC.path_db),
        )

        #
        # Start ClamD with our custom config
        #
        self.start_clamd(clamd_config=clamd_config)

        poll = self.proc.poll()
        assert poll == None  # subprocess is alive if poll() returns None

        # Check the big_file scan exceeds max filesize
        output = self.execute_command('{clamdscan} -c {clamd_config} --wait --ping 10 {test_exe}'.format(
            clamdscan=TC.clamdscan, clamd_config=clamd_config, test_exe=big_file))
        expected_results = ['MaxFileSize FOUND']
        unexpected_results = ['OK', 'MaxScanSize FOUND', 'Can\'t allocate memory ERROR']
        self.verify_output(output.out, expected=expected_results, unexpected=unexpected_results)
        assert output.ec == 1

        # Check the big_zip scan exceeds max scansize
        output = self.execute_command('{clamdscan} -c {clamd_config} {test_exe}'.format(
            clamdscan=TC.clamdscan, clamd_config=clamd_config, test_exe=big_zip))
        expected_results = ['MaxScanSize FOUND']
        unexpected_results = ['OK', 'MaxFileSize FOUND', 'Can\'t allocate memory ERROR']
        self.verify_output(output.out, expected=expected_results, unexpected=unexpected_results)
        assert output.ec == 1

    def test_clamd_12_onenote_disabled(self):
        self.step_name('Test that clamd.conf `ScanOneNote no` disables onenote support.')

        testpaths = [
            TC.path_build / "unit_tests" / "input" / "clamav_hdb_scanfiles" / "clam.exe.2007.one",
            TC.path_build / "unit_tests" / "input" / "clamav_hdb_scanfiles" / "clam.exe.2010.one",
            TC.path_build / "unit_tests" / "input" / "clamav_hdb_scanfiles" / "clam.exe.webapp-export.one",
        ]

        testfiles = ' '.join([str(testpath) for testpath in testpaths])

        # We'll use a config that sets `ScanOneNote yes`
        config = '''
            Foreground yes
            PidFile {pid}
            DatabaseDirectory {dbdir}
            LogFileMaxSize 0
            LogTime yes
            #Debug yes
            LogClean yes
            LogVerbose yes
            ExitOnOOM yes
            DetectPUA yes
            ScanPDF yes
            CommandReadTimeout 1
            MaxQueue 800
            MaxConnectionQueueLength 1024
            ScanOneNote yes
            '''.format(pid=TC.clamd_pid, dbdir=TC.path_db)
        if operating_system == 'windows':
            # Only have TCP socket option for Windows.
            config += '''
                TCPSocket {socket}
                TCPAddr localhost
                '''.format(socket=TC.clamd_port_num)
        else:
            # Use LocalSocket for Posix, because that's what check_clamd expects.
            config += '''
                LocalSocket {localsocket}
                TCPSocket {tcpsocket}
                TCPAddr localhost
                '''.format(localsocket=TC.clamd_socket, tcpsocket=TC.clamd_port_num)

        clamd_config = TC.path_tmp / 'clamd-test.conf'
        clamd_config.write_text(config)

        # Copy database to database path
        shutil.copy(
            str(TC.path_build / 'unit_tests' / 'input' / 'clamav.hdb'),
            str(TC.path_db),
        )

        #
        # Start ClamD with our custom config
        #
        self.start_clamd(clamd_config=clamd_config)

        poll = self.proc.poll()
        assert poll == None  # subprocess is alive if poll() returns None

        # Check the big_file scan exceeds max filesize
        output = self.execute_command('{clamdscan} -c {clamd_config} --wait --ping 10 {testfiles}'.format(
            clamdscan=TC.clamdscan, clamd_config=clamd_config, testfiles=testfiles))

        assert output.ec == 1  # virus found

        expected_results = ['{}: ClamAV-Test-File.UNOFFICIAL FOUND'.format(testpath.name) for testpath in testpaths]
        expected_results.append('Infected files: {}'.format(len(testpaths)))
        self.verify_output(output.out, expected=expected_results)


        #
        # Now retry with ScanOneNote disabled
        #

        # First kill clamd
        if self.proc != None:
            try:
                self.proc.terminate()
                self.proc.wait(timeout=120)
                self.proc.stdin.close()
            except OSError as exc:
                self.log.warning('Unexpected exception {}'.format(exc))
                pass  # ignore
            self.proc = None
        try:
            TC.clamd_pid.unlink()
        except Exception:
            pass # missing_ok=True is too for common use.
        try:
            TC.clamd_socket.unlink()
        except Exception:
            pass # missing_ok=True is too for common use.

        # Then update the config.
        # This time, we'll use a config that sets `ScanOneNote no`
        config = '''
            Foreground yes
            PidFile {pid}
            DatabaseDirectory {dbdir}
            LogFileMaxSize 0
            LogTime yes
            #Debug yes
            LogClean yes
            LogVerbose yes
            ExitOnOOM yes
            DetectPUA yes
            ScanPDF yes
            CommandReadTimeout 1
            MaxQueue 800
            MaxConnectionQueueLength 1024
            ScanOneNote no
            '''.format(pid=TC.clamd_pid, dbdir=TC.path_db)
        if operating_system == 'windows':
            # Only have TCP socket option for Windows.
            config += '''
                TCPSocket {socket}
                TCPAddr localhost
                '''.format(socket=TC.clamd_port_num)
        else:
            # Use LocalSocket for Posix, because that's what check_clamd expects.
            config += '''
                LocalSocket {localsocket}
                TCPSocket {tcpsocket}
                TCPAddr localhost
                '''.format(localsocket=TC.clamd_socket, tcpsocket=TC.clamd_port_num)

        clamd_config = TC.path_tmp / 'clamd-test.conf'
        clamd_config.write_text(config)

        #
        # Start ClamD with our custom config
        #
        self.start_clamd(clamd_config=clamd_config)

        poll = self.proc.poll()
        assert poll == None  # subprocess is alive if poll() returns None

        # Check the big_file scan exceeds max filesize
        output = self.execute_command('{clamdscan} -c {clamd_config} --wait --ping 10 {testfiles}'.format(
            clamdscan=TC.clamdscan, clamd_config=clamd_config, testfiles=testfiles))

        assert output.ec == 0  # virus found

        expected_results = ['{}: OK'.format(testpath.name) for testpath in testpaths]
        expected_results.append('Infected files: 0')
        self.verify_output(output.out, expected=expected_results)