infrastructure/extensions-website
4
8
Fork 10
Code Issues 20 Pull Requests 2 Projects Releases Wiki Activity

Scan files with clamdscan #77

Merged
Anna Sirota merged 17 commits from scan-file into main 2024-04-12 19:11:30 +02:00
Conversation 0 Commits 17 Files Changed 14 +8 -1
2 changed files with 8 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit d04ce7b3ad - Show all commits

4
files/tasks.py
View File

@ -1,3 +1,4 @@
import logging
import os.path import os.path
from background_task import background from background_task import background
@ -6,6 +7,8 @@ from django.conf import settings
import files.models import files.models
import files.utils import files.utils
logger = logging.getLogger(__name__)
@background() @background()
def scan(file_id: int): def scan(file_id: int):
@ -19,6 +22,7 @@ def scan(file_id: int):
'stderr': completed_process.stderr.decode(), 'stderr': completed_process.stderr.decode(),
'returncode': completed_process.returncode, 'returncode': completed_process.returncode,
} }
logger.info('File pk=%s scanned: exit code %s', file.pk, completed_process.returncode)
file_validation, is_new = files.models.FileValidation.objects.get_or_create( file_validation, is_new = files.models.FileValidation.objects.get_or_create(
file=file, defaults={'results': {completed_process.args[0]: scan_result}} file=file, defaults={'results': {completed_process.args[0]: scan_result}}
) )

5
files/utils.py
View File

@ -166,6 +166,9 @@ def guess_mimetype_from_content(file_obj) -> str:
def run_clamdscan(abs_path: str) -> 'subprocess.CompletedProcess': def run_clamdscan(abs_path: str) -> 'subprocess.CompletedProcess':
logger.info('Scanning file at path=%s', abs_path)
scan_args = ['clamdscan', '--fdpass', abs_path] scan_args = ['clamdscan', '--fdpass', abs_path]
logger.info('Running %s', scan_args) logger.info('Running %s', scan_args)
return subprocess.run(scan_args, capture_output=True) completed_process = subprocess.run(scan_args, capture_output=True)
logger.info('File at path=%s scanned: exit code %s', abs_path, completed_process.returncode)
return completed_process