Security Center:Detecção de arquivos maliciosos

package com.aliyun.filedetect.sample;

import java.io.File;
import java.util.HashMap;
import java.util.Map;

import com.aliyun.filedetect.*;

public class Sample {

	/**
	 * Synchronous file detection interface
	 * @param detector Detector object
	 * @param path Path of the file to be detected
	 * @param timeout_ms Set the timeout period in milliseconds
	 * @param wait_if_queuefull If the detection queue is full, false indicates returning an error immediately without waiting, while true indicates waiting until the queue has space.
	 * @throws InterruptedException
	 */
	public static DetectResult detectFileSync(OpenAPIDetector detector, String path, int timeout_ms, boolean wait_if_queuefull) throws InterruptedException {
		if (null == detector || null == path) return null;
		DetectResult result = null;
		while(true) {
			result = detector.detectSync(path, timeout_ms);
			if (null == result) break;
			if (result.error_code != ERR_CODE.ERR_DETECT_QUEUE_FULL) break;
			if (!wait_if_queuefull) break;
			detector.waitQueueAvailable(-1);
		}
		return result;
	}

	/**
	 * Asynchronous file detection interface
	 * @param detector Detector object
	 * @param path Path of the file to be detected
	 * @param timeout_ms Set the timeout period in milliseconds
	 * @param wait_if_queuefull If the detection queue is full, false indicates returning an error immediately without waiting, while true indicates waiting until the queue has space.
	 * @param callback Callback function for the result
	 * @throws InterruptedException
	 */
	public static int detectFile(OpenAPIDetector detector, String path, int timeout_ms, boolean wait_if_queuefull, IDetectResultCallback callback) throws InterruptedException {
		if (null == detector || null == path || null == callback) return ERR_CODE.ERR_INIT.value();
		int result = ERR_CODE.ERR_INIT.value();
		if (wait_if_queuefull) {
			final IDetectResultCallback real_callback = callback;
			callback = new IDetectResultCallback() {
				public void onScanResult(int seq, String file_path, DetectResult callback_res) {
					if (callback_res.error_code == ERR_CODE.ERR_DETECT_QUEUE_FULL) return;
					real_callback.onScanResult(seq, file_path, callback_res);
				}
			};
		}
		while(true) {
			result = detector.detect(path, timeout_ms, callback);
			if (result != ERR_CODE.ERR_DETECT_QUEUE_FULL.value()) break;
			if (!wait_if_queuefull) break;
			detector.waitQueueAvailable(-1);
		}
		return result;
	}

	/**
	 * Synchronous URL file detection interface
	 * @param detector Detector object
	 * @param url URL of the file to be detected
	 * @param md5 MD5 hash of the file to be detected
	 * @param timeout_ms Set the timeout period in milliseconds
	 * @param wait_if_queuefull If the detection queue is full, false indicates returning an error immediately without waiting, while true indicates waiting until the queue has space.
	 * @throws InterruptedException
	 */
	public static DetectResult detectUrlSync(OpenAPIDetector detector, String url, String md5, int timeout_ms, boolean wait_if_queuefull) throws InterruptedException {
		if (null == detector || null == url || null == md5) return null;
		DetectResult result = null;
		while(true) {
			result = detector.detectUrlSync(url, md5, timeout_ms);
			if (null == result) break;
			if (result.error_code != ERR_CODE.ERR_DETECT_QUEUE_FULL) break;
			if (!wait_if_queuefull) break;
			detector.waitQueueAvailable(-1);
		}
		return result;
	}

	/**
	 * Asynchronous URL file detection interface
	 * @param detector Detector object
	 * @param url URL of the file to be detected
	 * @param md5 MD5 hash of the file to be detected
	 * @param timeout_ms Set the timeout period in milliseconds
	 * @param wait_if_queuefull If the detection queue is full, false indicates returning an error immediately without waiting, while true indicates waiting until the queue has space.
	 * @param callback Callback function for the result
	 * @throws InterruptedException
	 */
	public static int detectUrl(OpenAPIDetector detector, String url, String md5, int timeout_ms, boolean wait_if_queuefull, IDetectResultCallback callback) throws InterruptedException {
		if (null == detector || null == url || null == md5 || null == callback) return ERR_CODE.ERR_INIT.value();
		int result = ERR_CODE.ERR_INIT.value();
		if (wait_if_queuefull) {
			final IDetectResultCallback real_callback = callback;
			callback = new IDetectResultCallback() {
				public void onScanResult(int seq, String file_path, DetectResult callback_res) {
					if (callback_res.error_code == ERR_CODE.ERR_DETECT_QUEUE_FULL) return;
					real_callback.onScanResult(seq, file_path, callback_res);
				}
			};
		}
		while(true) {
			result = detector.detectUrl(url, md5, timeout_ms, callback);
			if (result != ERR_CODE.ERR_DETECT_QUEUE_FULL.value()) break;
			if (!wait_if_queuefull) break;
			detector.waitQueueAvailable(-1);
		}
		return result;
	}

	/**
	 * Format the detection result
	 * @param result Detection result object
	 * @return Formatted string
	 */
	public static String formatDetectResult(DetectResult result) {
		if (result.isSucc()) {
			DetectResult.DetectResultInfo info = result.getDetectResultInfo();
			String msg = String.format("[DETECT RESULT] [SUCCEED] %s", formatDetectResultInfo(info));
			if (info.compresslist != null) {
				int idx = 1;
				for (DetectResult.CompressFileDetectResultInfo comp_res : info.compresslist) {
					msg += String.format("\n\t\t\t [COMPRESS FILE] [IDX:%d] %s", idx++, formatCompressFileDetectResultInfo(comp_res));
				}
			}
			return msg;
		}
		DetectResult.ErrorInfo info = result.getErrorInfo();
		return String.format("[DETECT RESULT] [FAIL] md5: %s, time: %d, error_code: %s, error_message: %s"
				, info.md5, info.time, info.error_code.name(), info.error_string);
	}

	private static String formatDetectResultInfo(DetectResult.DetectResultInfo info) {
		String msg = String.format("MD5: %s, TIME: %d, RESULT: %s, SCORE: %d", info.md5, info.time, info.result.name(), info.score);
		if (info.compresslist != null) {
			msg += String.format(", COMPRESS_FILES: %d", info.compresslist.size());
		}
		DetectResult.VirusInfo vinfo = info.getVirusInfo();
		if (vinfo != null) {
			msg += String.format(", VIRUS_TYPE: %s, EXT_INFO: %s", vinfo.virus_type, vinfo.ext_info);
		}
		return msg;
	}
	private static String formatCompressFileDetectResultInfo(DetectResult.CompressFileDetectResultInfo info) {
		String msg = String.format("PATH: %s, \t\t RESULT: %s, SCORE: %d", info.path, info.result.name(), info.score);
		DetectResult.VirusInfo vinfo = info.getVirusInfo();
		if (vinfo != null) {
			msg += String.format(", VIRUS_TYPE: %s, EXT_INFO: %s", vinfo.virus_type, vinfo.ext_info);
		}
		return msg;
	}

	/**
	 * Synchronously detect a directory or file
	 * @param path The specified path can be a file or a directory. If it is a directory, its content is recursively traversed.
	 * @param is_sync Specifies whether to use the synchronous interface. We recommend using the asynchronous interface. true: synchronous, false: asynchronous
	 * @throws InterruptedException
	 */
	public static void detectDirOrFileSync(OpenAPIDetector detector, String path, int timeout_ms, Map<String, DetectResult> result_map) throws InterruptedException {
		File file = new File(path);
		String abs_path = file.getAbsolutePath();
		if (file.isDirectory()) {
			String[] ss = file.list();
	        if (ss == null) return;
	        for (String s : ss) {
	        	String subpath = abs_path + File.separator + s;
	        	detectDirOrFileSync(detector, subpath, timeout_ms, result_map);
	        }
			return;
		}

    	System.out.println(String.format("[detectFileSync] [BEGIN] queueSize: %d, path: %s, timeout: %d", detector.getQueueSize(), abs_path, timeout_ms));
		DetectResult res = detectFileSync(detector, abs_path, timeout_ms, true);
    	System.err.println(String.format("                 [ END ] %s", formatDetectResult(res)));
		result_map.put(abs_path, res);
	}

	/**
	 * Asynchronously detect a directory or file
	 * @param path The specified path can be a file or a directory. If it is a directory, its content is recursively traversed.
	 * @param is_sync Specifies whether to use the synchronous interface. We recommend using the asynchronous interface. true: synchronous, false: asynchronous
	 * @throws InterruptedException
	 */
	public static void detectDirOrFile(OpenAPIDetector detector, String path, int timeout_ms, IDetectResultCallback callback) throws InterruptedException {
		File file = new File(path);
		String abs_path = file.getAbsolutePath();
		if (file.isDirectory()) {
			String[] ss = file.list();
	        if (ss == null) return;
	        for (String s : ss) {
	        	String subpath = abs_path + File.separator + s;
	        	detectDirOrFile(detector, subpath, timeout_ms, callback);
	        }
	        return;
		}

		int seq = detectFile(detector, abs_path, timeout_ms, true, callback);
		System.out.println(String.format("[detectFile] [BEGIN] seq: %d, queueSize: %d, path: %s, timeout: %d", seq, detector.getQueueSize(), abs_path, timeout_ms));
	}

	/**
	 * Start detecting a file or directory
	 * @param path The specified path can be a file or a directory. If it is a directory, its content is recursively traversed.
	 * @param is_sync Specifies whether to use the synchronous interface. We recommend using the asynchronous interface. true: synchronous, false: asynchronous
	 * @throws InterruptedException
	 */
	public static void scan(final OpenAPIDetector detector, String path, int detect_timeout_ms, boolean is_sync) throws InterruptedException {
		System.out.println(String.format("[SCAN] [START] path: %s, detect_timeout_ms: %d, is_sync: %b", path, detect_timeout_ms, is_sync));
		long start_time = System.currentTimeMillis();
		final Map<String, DetectResult> result_map = new HashMap<>();
		if (is_sync) {
			detectDirOrFileSync(detector, path, detect_timeout_ms, result_map);
		} else {
			detectDirOrFile(detector, path, detect_timeout_ms, new IDetectResultCallback() {
				public void onScanResult(int seq, String file_path, DetectResult callback_res) {
			    	System.err.println(String.format("[detectFile] [ END ] seq: %d, queueSize: %d, %s", seq, detector.getQueueSize(), formatDetectResult(callback_res)));
					result_map.put(file_path, callback_res);
				}
			});
			// Wait for the task to complete.
			detector.waitQueueEmpty(-1);
		}
		long used_time = System.currentTimeMillis() - start_time;
		System.out.println(String.format("[SCAN] [ END ] used_time: %d, files: %d", used_time, result_map.size()));

		int fail_count = 0;
		int white_count = 0;
		int black_count = 0;
		for (Map.Entry<String, DetectResult> entry : result_map.entrySet()) {
			DetectResult res = entry.getValue();
			if (res.isSucc()) {
				if (res.getDetectResultInfo().result == DetectResult.RESULT.RES_BLACK) {
					black_count ++;
				} else {
					white_count ++;
				}
			} else {
				fail_count ++;
			}
		}
		System.out.println(String.format("             fail_count: %d, white_count: %d, black_count: %d"
				, fail_count, white_count, black_count));
	}

    public static void main(String[] args_) throws Exception {
    	// Get the detector instance.
    	OpenAPIDetector detector = OpenAPIDetector.getInstance();

    	// Initialize.
    	ERR_CODE init_ret = detector.init(System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
    	System.out.println("INIT RET: " + init_ret.name());

    	// Set decompression parameters (optional, compressed packages are not decompressed by default).
    	boolean decompress = true; // Specifies whether to identify and decompress compressed files. Default is false.
    	int decompressMaxLayer = 5; // Maximum decompression level. Effective when decompress is true.
    	int decompressMaxFileCount = 1000; // Maximum number of decompressed files. Effective when decompress is true.
    	ERR_CODE initdec_ret = detector.initDecompress(decompress, decompressMaxLayer, decompressMaxFileCount);
    	System.out.println("INIT_DECOMPRESS RET: " + initdec_ret.name());

    	if (true) {
    		// Example 1: Scan a local directory or file.
    		boolean is_sync_scan = false; // Specifies whether to use asynchronous or synchronous detection. Asynchronous detection offers better performance. false indicates asynchronous detection.
        	int timeout_ms = 500000;  // Detection time for a single sample, in milliseconds.
        	String path = "test2.php"; // The file or directory to be scanned.
        	// Start the scan and wait for it to finish.
        	scan(detector, path, timeout_ms, is_sync_scan);
    	}

    	if (true) {
    		// Example 2: Scan a URL file.
        	int timeout_ms = 500000;  // Detection time for a single sample, in milliseconds.
        	String url = "https://xxxxxxxx.oss-cn-hangzhou-1.aliyuncs.com/xxxxx/xxxxxxxxxxxxxx?Expires=1*****25&OSSAccessKeyId=xxx"; // The URL file to be scanned.
        	String md5 = "a767f*************6e21d000000"; // The MD5 hash of the file to be scanned.
        	// Synchronous scan. To perform an asynchronous scan, call the detectUrl interface.
        	System.out.println(String.format("[detectUrlSync] [BEGIN] URL: %s, MD5: %s, TIMEOUT: %d", url, md5, timeout_ms));
        	DetectResult result = detectUrlSync(detector, url, md5, timeout_ms, true);
        	System.err.println(String.format("[detectUrlSync] [ END ] %s", formatDetectResult(result)));
    	}

		// Deinitialize.
		System.out.println("Over.");
    	detector.uninit();
    }
}

# -*- coding: utf-8 -*-
import os
import sys
from typing import List
import threading
import time
import traceback

from alibabacloud_filedetect.OpenAPIDetector import OpenAPIDetector
from alibabacloud_filedetect.IDetectResultCallback import IDetectResultCallback
from alibabacloud_filedetect.ERR_CODE import ERR_CODE
from alibabacloud_filedetect.DetectResult import DetectResult

class Sample(object):

    def __init__(self):
        pass

    """
    Synchronous file detection interface
    @param detector: Detector object
    @param path: Path of the file to be detected
    @param timeout_ms: Set the timeout period in milliseconds
    @param wait_if_queuefull: If the detection queue is full, False indicates returning an error immediately without waiting, while True indicates waiting until the queue has space.
    """
    def detectFileSync(self, detector, path, timeout_ms, wait_if_queuefull):
        if detector is None or path is None:
            return None
        result = None
        while True:
            result = detector.detectSync(path, timeout_ms)
            if result is None:
                break
            if result.error_code != ERR_CODE.ERR_DETECT_QUEUE_FULL:
                break
            if wait_if_queuefull is False:
                break
            detector.waitQueueAvailable(-1)
        return result

    """
    Asynchronous file detection interface
    @param detector: Detector object
    @param path: Path of the file to be detected
    @param timeout_ms: Set the timeout period in milliseconds
    @param wait_if_queuefull: If the detection queue is full, False indicates returning an error immediately without waiting, while True indicates waiting until the queue has space.
    @param callback: Callback function for the result
    """
    def detectFile(self, detector, path, timeout_ms, wait_if_queuefull, callback):
        if detector is None or path is None or callback is None:
            return ERR_CODE.ERR_INIT.value
        result = ERR_CODE.ERR_INIT.value
        if wait_if_queuefull:
            real_callback = callback
            class AsyncTaskCallback(IDetectResultCallback):
                def onScanResult(self, seq, file_path, callback_res):
                    if callback_res.error_code == ERR_CODE.ERR_DETECT_QUEUE_FULL:
                        return
                    real_callback.onScanResult(seq, file_path, callback_res)
            callback = AsyncTaskCallback()
        while True:
            result = detector.detect(path, timeout_ms, callback)
            if result != ERR_CODE.ERR_DETECT_QUEUE_FULL.value:
                break
            if wait_if_queuefull is False:
                break
            detector.waitQueueAvailable(-1)
        return result

    """
    Synchronous URL file detection interface
    @param detector: Detector object
    @param url: URL of the file to be detected
    @param md5: MD5 hash of the file to be detected
    @param timeout_ms: Set the timeout period in milliseconds
    @param wait_if_queuefull: If the detection queue is full, False indicates returning an error immediately without waiting, while True indicates waiting until the queue has space.
    """
    def detectUrlSync(self, detector, url, md5, timeout_ms, wait_if_queuefull):
        if detector is None or url is None or md5 is None:
            return None
        result = None
        while True:
            result = detector.detectUrlSync(url, md5, timeout_ms)
            if result is None:
                break
            if result.error_code != ERR_CODE.ERR_DETECT_QUEUE_FULL:
                break
            if wait_if_queuefull is False:
                break
            detector.waitQueueAvailable(-1)
        return result

    """
    Asynchronous URL file detection interface
    @param detector: Detector object
    @param url: URL of the file to be detected
    @param md5: MD5 hash of the file to be detected
    @param timeout_ms: Set the timeout period in milliseconds
    @param wait_if_queuefull: If the detection queue is full, False indicates returning an error immediately without waiting, while True indicates waiting until the queue has space.
    @param callback: Callback function for the result
    """
    def detectUrl(self, detector, url, md5, timeout_ms, wait_if_queuefull, callback):
        if detector is None or url is None or md5 is None or callback is None:
            return ERR_CODE.ERR_INIT.value
        result = ERR_CODE.ERR_INIT.value
        if wait_if_queuefull:
            real_callback = callback
            class AsyncTaskCallback(IDetectResultCallback):
                def onScanResult(self, seq, file_path, callback_res):
                    if callback_res.error_code == ERR_CODE.ERR_DETECT_QUEUE_FULL:
                        return
                    real_callback.onScanResult(seq, file_path, callback_res)
            callback = AsyncTaskCallback()
        while True:
            result = detector.detectUrl(url, md5, timeout_ms, callback)
            if result != ERR_CODE.ERR_DETECT_QUEUE_FULL.value:
                break
            if wait_if_queuefull is False:
                break
            detector.waitQueueAvailable(-1)
        return result

    """
    Format the detection result
    @param result: Detection result object
    @return: Formatted string
    """
    @staticmethod
    def formatDetectResult(result):
        msg = ""
        if result.isSucc():
            info = result.getDetectResultInfo()
            msg = "[DETECT RESULT] [SUCCEED] {}".format(Sample.formatDetectResultInfo(info))
            if info.compresslist is not None:
                idx = 1
                for comp_res in info.compresslist:
                    msg += "\n\t\t\t [COMPRESS FILE] [IDX:{}] {}".format(idx, Sample.formatCompressFileDetectResultInfo(comp_res))
                    idx += 1
        else:
            info = result.getErrorInfo()
            msg = "[DETECT RESULT] [FAIL] md5: {}, time: {}, error_code: {}, error_message: {}".format(info.md5,
                info.time, info.error_code.name, info.error_string)
        return msg

    @staticmethod
    def formatDetectResultInfo(info):
        msg = "MD5: {}, TIME: {}, RESULT: {}, SCORE: {}".format(info.md5, info.time, info.result.name, info.score)
        if info.compresslist is not None:
            msg += ", COMPRESS_FILES: {}".format(len(info.compresslist))
        vinfo = info.getVirusInfo()
        if vinfo is not None:
            msg += ", VIRUS_TYPE: {}, EXT_INFO: {}".format(vinfo.virus_type, vinfo.ext_info)
        return msg

    @staticmethod
    def formatCompressFileDetectResultInfo(info):
        msg = "PATH: {}, \t\t RESULT: {}, SCORE: {}".format(info.path, info.result.name, info.score)
        vinfo = info.getVirusInfo()
        if vinfo is not None:
            msg += ", VIRUS_TYPE: {}, EXT_INFO: {}".format(vinfo.virus_type, vinfo.ext_info)
        return msg

    """
    Synchronously detect a directory or file
    @param path: The specified path can be a file or a directory. If it is a directory, its content is recursively traversed.
    @param is_sync: Specifies whether to use the synchronous interface. We recommend using the asynchronous interface. True: synchronous, False: asynchronous
    """
    def detectDirOrFileSync(self, detector, path, timeout_ms, result_map):
        abs_path = os.path.abspath(path)
        if os.path.isdir(abs_path):
            sub_files = os.listdir(abs_path)
            if len(sub_files) == 0:
                return
            for sub_file in sub_files:
                sub_path = os.path.join(abs_path, sub_file)
                self.detectDirOrFileSync(detector, sub_path, timeout_ms, result_map)
            return

        print("[detectFileSync] [BEGIN] queueSize: {}, path: {}, timeout: {}".format(
            detector.getQueueSize(), abs_path, timeout_ms))
        res = self.detectFileSync(detector, abs_path, timeout_ms, True)
        print("                 [ END ] {}".format(Sample.formatDetectResult(res)))
        result_map[abs_path] = res
        return

    """
    Asynchronously detect a directory or file
    @param path: The specified path can be a file or a directory. If it is a directory, its content is recursively traversed.
    @param is_sync: Specifies whether to use the synchronous interface. We recommend using the asynchronous interface. True: synchronous, False: asynchronous
    """
    def detectDirOrFile(self, detector, path, timeout_ms, callback):
        abs_path = os.path.abspath(path)
        if os.path.isdir(abs_path):
            sub_files = os.listdir(abs_path)
            if len(sub_files) == 0:
                return
            for sub_file in sub_files:
                sub_path = os.path.join(abs_path, sub_file)
                self.detectDirOrFile(detector, sub_path, timeout_ms, callback)
            return

        seq = self.detectFile(detector, abs_path, timeout_ms, True, callback)
        print("[detectFile] [BEGIN] seq: {}, queueSize: {}, path: {}, timeout: {}".format(
            seq, detector.getQueueSize(), abs_path, timeout_ms))
        return

    """
    Start detecting a file or directory
    @param path: The specified path can be a file or a directory. If it is a directory, its content is recursively traversed.
    @param is_sync: Specifies whether to use the synchronous interface. We recommend using the asynchronous interface. True: synchronous, False: asynchronous
    """
    def scan(self, detector, path, detect_timeout_ms, is_sync):
        try:
            print("[SCAN] [START] path: {}, detect_timeout_ms: {}, is_sync: {}".format(path, detect_timeout_ms, is_sync))
            start_time = time.time()
            result_map = {}
            if is_sync:
                self.detectDirOrFileSync(detector, path, detect_timeout_ms, result_map)
            else:
                class AsyncTaskCallback(IDetectResultCallback):
                    def onScanResult(self, seq, file_path, callback_res):
                        print("[detectFile] [ END ] seq: {}, queueSize: {}, {}".format(seq,
                            detector.getQueueSize(), Sample.formatDetectResult(callback_res)))
                        result_map[file_path] = callback_res
                self.detectDirOrFile(detector, path, detect_timeout_ms, AsyncTaskCallback())
                # Wait for the task to complete.
                detector.waitQueueEmpty(-1)

            used_time_ms = (time.time() - start_time) * 1000
            print("[SCAN] [ END ] used_time: {}, files: {}".format(int(used_time_ms), len(result_map)))

            failed_count = 0
            white_count = 0
            black_count = 0
            for file_path, res in result_map.items():
                if res.isSucc():
                    if res.getDetectResultInfo().result == DetectResult.RESULT.RES_BLACK:
                        black_count += 1
                    else:
                        white_count += 1
                else:
                    failed_count += 1

            print("               fail_count: {}, white_count: {}, black_count: {}".format(
                failed_count, white_count, black_count))

        except Exception as e:
            print(traceback.format_exc(), file=sys.stderr)

    def main(self):

        # Get the detector instance.
        detector = OpenAPIDetector.get_instance()

        # Read the AccessKey ID and AccessKey Secret from environment variables.
        access_key_id = os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID')
        access_key_secret = os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET')

        # Initialize.
        init_ret = detector.init(access_key_id, access_key_secret)
        print("INIT RET: {}".format(init_ret.name))

        # Set decompression parameters (optional, compressed packages are not decompressed by default).
        decompress = True # Specifies whether to identify and decompress compressed files. Default is false.
        decompressMaxLayer = 5 # Maximum decompression level. Effective when decompress is true.
        decompressMaxFileCount = 1000 # Maximum number of decompressed files. Effective when decompress is true.
        initdec_ret = detector.initDecompress(decompress, decompressMaxLayer, decompressMaxFileCount)
        print("INIT_DECOMPRESS RET: {}".format(initdec_ret.name))

        if True:
            # Example 1: Scan a local directory or file.
            is_sync_scan = False # Specifies whether to use asynchronous or synchronous detection. Asynchronous detection offers better performance. False indicates asynchronous detection.
            timeout_ms = 500000 # Detection time for a single sample, in milliseconds.
            path = "test.bin" # The file or directory to be scanned.
            # Start the scan and wait for it to finish.
            self.scan(detector, path, timeout_ms, is_sync_scan)

        if True:
            # Example 2: Scan a URL file.
            timeout_ms = 500000
            url = "https://xxxxxxxx.oss-cn-hangzhou-1.aliyuncs.com/xxxxx/xxxxxxxxxxxxxx?Expires=1671448125&OSSAccessKeyId=xxx" # The URL file to be scanned.
            md5 = "a767ffc59d93125c7505b6e21d000000"
            # Synchronous scan. To perform an asynchronous scan, call the detectUrl interface.
            print("[detectUrlSync] [BEGIN] URL: {}, MD5: {}, TIMEOUT: {}".format(url, md5, timeout_ms))
            result = self.detectUrlSync(detector, url, md5, timeout_ms, True)
            print("[detectUrlSync] [ END ] {}".format(Sample.formatDetectResult(result)))

        # Deinitialize.
        print("Over.")
        detector.uninit()

if __name__ == "__main__":
    sample = Sample()
    sample.main()