How to Scan Files Before Uploading Them to Amazon S3 in Python

Cloudmersive

2 min read

·

2 days ago

--

Listen

Share

Our pipeline might push files to S3 and move on, but the upload is rarely the end of their journey. Other systems tend to read files from the buckets we fill; lambdas fire on new objects, nightly jobs process everything inside, and other services download files on demand. If there’s a malicious file sitting in a bucket, it’s a problem for every one of those systems.

Press enter or click to view image in full size

While it’s true that we could scan files with an S3 trigger after upload, that would still let a bad file land in our bucket first — and we don’t want that. Scanning before the upload means the file never arrives at all, and that’s a much safer bet.

Let’s handle that with a pre-upload gate. We’ll use boto3 and the Cloudmersive Advanced Virus Scan API.

Installing the dependencies

pip install boto3 cloudmersive-virus-api-client

Connecting to boto3 and Cloudmersive

import os
import boto3
import cloudmersive_virus_api_client

cfg = cloudmersive_virus_api_client.Configuration()
cfg.api_key["Apikey"] = os.environ["CLOUDMERSIVE_API_KEY"]

scan_api = cloudmersive_virus_api_client.ScanApi(
    cloudmersive_virus_api_client.ApiClient(cfg)
)

s3 = boto3.client("s3")
BUCKET = "my-upload-bucket"

Note that we can get a free Cloudmersive API key by creating a free account on the Cloudmersive website.

Building a gate for a single file

Since these files already sit on disk, we can scan each file path directly. We’ll just put the whole decision (scan → upload || refuse) into one function that reports back what it did.

import pathlib

def upload_if_clean(path: pathlib.Path) -> str:
    verdict = scan_api.scan_file_advanced(
        str(path),
        allow_executables=False,
        allow_scripts=False,
        allow_macros=False,
        allow_invalid_files=False,
        allow_password_protected_files=False,
        allow_unsafe_archives=False,
    )

    if verdict.clean_result is not True:
        return "blocked"

    s3.upload_file(str(path), BUCKET, path.name)
    return "uploaded"p

Running it over a folder

Now that we’ve implemented per-file decision making, processing a directory is a short loop. Each file is responsible for announcing its own outcome:

for item in pathlib.Path("./outbox").iterdir():
    if item.is_file():
        print(f"{item.name}: {upload_if_clean(item)}")

A blocked file simply can’t land in the bucket because it never reaches upload_file.

Wrapping up

Vetting files in this simple, easy way keeps our bucket safe by design instead of relying on a cleanup pass later. The same gate works against any bucket or prefix we aim it at because the scan and transfer stay decoupled.