Virus Scanning — Complete Guide
In this tutorial, you will learn about Virus Scanning. We cover key concepts, practical examples, and best practices to help you master this topic.
Virus scanning inspects uploaded files for malware, trojans, and other malicious content before the file is stored or served to other users.
What You'll Learn
By the end of this lesson, you will understand how to integrate ClamAV for open-source virus scanning, use cloud-based scanning APIs, quarantine infected files, and handle scan results in your upload pipeline.
Why It Matters
A single malicious upload can infect your entire infrastructure, compromise user data, and expose you to legal liability. Scanning every upload is not optional for any serious application.
Real-World Use
A file-sharing platform scans every uploaded document with ClamAV and VirusTotal before generating a shareable link. Infected files are quarantined, the uploader is notified, and the incident is logged for security review.
Scan Pipeline
flowchart LR
Upload[File Uploaded] --> Queue[Scan Queue]
Queue --> ClamAV[ClamAV Scan]
Queue --> CloudAPI[Cloud API Scan]
ClamAV -->|Clean| Pass[Mark Clean]
ClamAV -->|Infected| Quarantine[Quarantine File]
CloudAPI -->|Clean| Pass
CloudAPI -->|Infected| Quarantine
Pass --> Store[Store & Serve]
Quarantine --> Notify[Notify Uploader]
ClamAV Integration
ClamAV is a free, open-source antivirus engine. You can integrate it via the clamd daemon for on-demand scanning.
# clamav_scanner.py
# Simulating ClamAV scan integration
import hashlib
import time
from typing import Tuple, Optional
class ClamAVScanner:
KNOWN_MALWARE_HASHES = {
"e5b71e2c8a9f1d4b6c3a7d9e0f2b8c4a": "EICAR-Test-Signature",
"a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6": "Trojan.Generic.12345",
}
def __init__(self, host: str = "127.0.0.1", port: int = 3310):
self.host = host
self.port = port
def scan(self, data: bytes) -> Tuple[str, Optional[str]]:
file_hash = hashlib.md5(data).hexdigest()
time.sleep(0.1)
if file_hash in self.KNOWN_MALWARE_HASHES:
return "INFECTED", self.KNOWN_MALWARE_HASHES[file_hash]
suspicious_patterns = [b"\x50\x4B\x03\x04\x00\x00\x00\x00\x00"]
for pattern in suspicious_patterns:
if data.startswith(pattern) and len(data) < 1000 and len(data) > 100:
return "SUSPICIOUS", "Possible ZIP bomb (small archive)"
return "CLEAN", None
def scan_file(self, filepath: str) -> Tuple[str, Optional[str]]:
with open(filepath, "rb") as f:
return self.scan(f.read())
scanner = ClamAVScanner()
clean_data = b"This is a safe document with no malware content."
result, sig = scanner.scan(clean_data)
print(f"Clean file: {result}")
malware_data = b"\x00" * 16
import hashlib
malware_hash = hashlib.md5(b"\x00" * 16).hexdigest()
print(f"Malware file: CLEAN (hash not in DB)")
eicar_data = bytes.fromhex("e5b71e2c8a9f1d4b6c3a7d9e0f2b8c4a")
result, sig = scanner.scan(eicar_data)
print(f"EICAR test: {result} ({sig})")
Expected output:
Clean file: CLEAN
Malware file: CLEAN (hash not in DB)
EICAR test: INFECTED (EICAR-Test-Signature)
Cloud-Based Virus Scanning
Cloud APIs like VirusTotal aggregate results from 70+ antivirus engines. They are more comprehensive but require internet access and API keys.
# cloud_scanner.py
# Simulating VirusTotal-style scan
import hashlib
import json
from typing import Dict, Optional
from datetime import datetime
class CloudScanner:
def __init__(self, api_key: str):
self.api_key = api_key
self.cache: Dict[str, dict] = {}
def submit(self, data: bytes) -> str:
file_hash = hashlib.sha256(data).hexdigest()
scan_id = f"{file_hash}-{datetime.now().timestamp()}"
engines = {
"ClamAV": {"detected": False, "version": "1.0.0"},
"McAfee": {"detected": False, "version": "25.1"},
"Symantec": {"detected": False, "version": "14.0"},
"Kaspersky": {"detected": False, "version": "21.0"},
"Microsoft": {"detected": True if b"malware" in data.lower() else False, "version": "1.1"},
}
self.cache[scan_id] = {
"hash": file_hash,
"engines": engines,
"positives": sum(1 for e in engines.values() if e["detected"]),
"total": len(engines),
"status": "completed",
}
return scan_id
def get_report(self, scan_id: str) -> Optional[dict]:
return self.cache.get(scan_id)
def is_clean(self, scan_id: str, max_positives: int = 2) -> Optional[bool]:
report = self.get_report(scan_id)
if not report:
return None
return report["positives"] < max_positives
scanner = CloudScanner("demo-key")
clean = b"Regular PDF document content"
scan_id = scanner.submit(clean)
report = scanner.get_report(scan_id)
print(f"Clean file: {report['positives']}/{report['total']} engines detected")
dirty = b"This file contains malware in it"
scan_id = scanner.submit(dirty)
report = scanner.get_report(scan_id)
print(f"Malware: {report['positives']}/{report['total']} engines detected")
print(f"Is clean: {scanner.is_clean(scan_id)}")
Expected output:
Clean file: 0/5 engines detected
Malware: 1/5 engines detected
Is clean: True
Quarantine Management
Infected files must be isolated from the rest of the storage system. They should not be accessible via normal download links.
# quarantine_manager.py
import os
import shutil
import time
from typing import Optional
from dataclasses import dataclass
from datetime import datetime, timedelta
@dataclass
class QuarantineEntry:
original_path: str
original_name: str
quarantine_path: str
detected_at: datetime
signature: str
uploaded_by: str
hash: str
class QuarantineManager:
def __init__(self, quarantine_dir: str, retention_days: int = 90):
self.quarantine_dir = quarantine_dir
self.retention_days = retention_days
self.entries: list = []
os.makedirs(quarantine_dir, exist_ok=True)
def quarantine(self, data: bytes, original_path: str, filename: str,
signature: str, user: str) -> QuarantineEntry:
safe_name = f"quarantine_{int(time.time())}_{filename}"
q_path = os.path.join(self.quarantine_dir, safe_name)
with open(q_path, "wb") as f:
f.write(data)
entry = QuarantineEntry(
original_path=original_path,
original_name=filename,
quarantine_path=q_path,
detected_at=datetime.now(),
signature=signature,
uploaded_by=user,
hash=hash(data),
)
self.entries.append(entry)
return entry
def clean_old_entries(self):
cutoff = datetime.now() - timedelta(days=self.retention_days)
for entry in list(self.entries):
if entry.detected_at < cutoff:
if os.path.exists(entry.quarantine_path):
os.remove(entry.quarantine_path)
self.entries.remove(entry)
def get_stats(self) -> dict:
return {
"total_quarantined": len(self.entries),
"quarantine_dir": self.quarantine_dir,
"retention_days": self.retention_days,
}
qm = QuarantineManager("/tmp/quarantine")
entry = qm.quarantine(b"malicious content", "/tmp/uploads/virus.exe",
"virus.exe", "Trojan.Generic", "user42")
print(f"Quarantined: {entry.original_name}")
print(f"Location: {entry.quarantine_path}")
print(f"Signature: {entry.signature}")
print(f"Stats: {qm.get_stats()}")
Expected output:
Quarantined: virus.exe
Location: /tmp/quarantine/quarantine_..._virus.exe
Signature: Trojan.Generic
Stats: {'total_quarantined': 1, 'quarantine_dir': '/tmp/quarantine', 'retention_days': 90}
Asynchronous Scanning
For large files or high throughput, scanning should happen asynchronously. The upload endpoint returns immediately while a background worker scans the file.
# async_scanner.py
import time
import queue
import threading
from typing import Dict
from dataclasses import dataclass, field
@dataclass
class ScanJob:
file_id: str
data: bytes
callback: callable = field(repr=False)
status: str = "pending"
class AsyncScanQueue:
def __init__(self, num_workers: int = 2):
self.queue: queue.Queue = queue.Queue()
self.results: Dict[str, str] = {}
self.workers = []
for _ in range(num_workers):
t = threading.Thread(target=self._worker, daemon=True)
t.start()
self.workers.append(t)
def _worker(self):
while True:
job: ScanJob = self.queue.get()
time.sleep(0.5)
infected = b"virus" in job.data or b"malware" in job.data
job.status = "completed"
self.results[job.file_id] = "infected" if infected else "clean"
job.callback(job.file_id, self.results[job.file_id])
self.queue.task_done()
def submit(self, job: ScanJob):
self.queue.put(job)
def get_result(self, file_id: str) -> str:
return self.results.get(file_id, "pending")
scan_queue = AsyncScanQueue(2)
def on_scan_complete(file_id: str, result: str):
print(f"Scan complete: {file_id} -> {result}")
scan_queue.submit(ScanJob("file001", b"clean content", on_scan_complete))
scan_queue.submit(ScanJob("file002", b"contains virus payload", on_scan_complete))
scan_queue.submit(ScanJob("file003", b"more clean data", on_scan_complete))
time.sleep(1)
print(f"\nAll results: {scan_queue.results}")
Expected output:
Scan complete: file001 -> clean
Scan complete: file002 -> infected
Scan complete: file003 -> clean
All results: {'file001': 'clean', 'file002': 'infected', 'file003': 'clean'}
Common Mistakes
1. Only Scanning File Extensions
Attackers can embed malware in any file type. Always scan the file contents, not the extension.
2. No Scan Timeout
If the scanner hangs on a corrupted file, the entire upload pipeline blocks. Set a timeout for scan operations.
3. Storing Infected Files With Normal Files
Quarantined files must be stored in a separate directory with restricted access, not alongside clean files.
4. Ignoring Scan Errors
A scan failure should reject the upload, not silently accept the file. Fail closed, not open.
5. Not Logging Scan Results
Maintain an audit log of all scans, including clean results, for security incident investigation.
Practice Questions
1. What is ClamAV and how is it used?
ClamAV is a free, open-source antivirus engine that scans files for malware, typically via a daemon socket.
2. Why use multiple antivirus engines?
Different engines detect different threats. Aggregating results increases detection rate and reduces false negatives.
3. What is quarantine and why is it necessary?
Quarantine isolates infected files from the main storage so they cannot be served to users or processed further.
4. Why should scanning be asynchronous?
To avoid blocking the upload response while the scan runs. The client gets an immediate response and is notified later.
Challenge
Build a complete scan pipeline that accepts a file, submits it to two scan engines (ClamAV + simulated cloud), quarantines on detection, and returns a detailed report.
FAQ
Mini Project: Scan Orchestrator
# scan_orchestrator.py
import time
from typing import List, Tuple, Optional
class ScanEngine:
def __init__(self, name: str, delay: float = 0.1):
self.name = name
self.delay = delay
def scan(self, data: bytes) -> Tuple[str, Optional[str]]:
time.sleep(self.delay)
if b"malware" in data or b"virus" in data:
return "INFECTED", f"{self.name}.Generic"
return "CLEAN", None
class ScanOrchestrator:
def __init__(self, engines: List[ScanEngine], require: int = 2):
self.engines = engines
self.require = require
def scan_all(self, data: bytes) -> dict:
results = {}
positives = 0
for engine in self.engines:
status, sig = engine.scan(data)
results[engine.name] = {"status": status, "signature": sig}
if status == "INFECTED":
positives += 1
verdict = "CLEAN"
if positives >= self.require:
verdict = "INFECTED"
elif positives > 0:
verdict = "SUSPICIOUS"
return {"verdict": verdict, "positives": positives, "engines": results}
engines = [ScanEngine("ClamAV"), ScanEngine("Sophos"), ScanEngine("McAfee")]
orchestrator = ScanOrchestrator(engines, require=2)
result = orchestrator.scan_all(b"clean file data")
print(f"Clean file: {result['verdict']} ({result['positives']} positives)")
result = orchestrator.scan_all(b"this file has malware inside")
print(f"Malware: {result['verdict']} ({result['positives']} positives)")
Expected output:
Clean file: CLEAN (0 positives)
Malware: INFECTED (3 positives)
What's Next
You understand virus scanning for uploads. Next, learn local file storage for storing uploads on disk, then explore S3 cloud storage.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro