Add support for ClamAV

0x0-vscan.service

@@ -0,0 +1,22 @@
+[Unit]
+Description=Scan 0x0 files with ClamAV
+After=remote-fs.target clamd.service
+
+[Service]
+Type=oneshot
+User=nullptr
+WorkingDirectory=/path/to/0x0
+BindPaths=/path/to/0x0
+
+Environment=FLASK_APP=fhost
+ExecStart=/usr/bin/flask vscan
+ProtectProc=noaccess
+ProtectSystem=strict
+ProtectHome=tmpfs
+PrivateTmp=true
+PrivateUsers=true
+ProtectKernelLogs=true
+LockPersonality=true
+
+[Install]
+WantedBy=multi-user.target

0x0-vscan.timer

@@ -0,0 +1,9 @@
+[Unit]
+Description=Scan 0x0 files with ClamAV
+
+[Timer]
+OnCalendar=hourly
+Persistent=true
+
+[Install]
+WantedBy=timers.target

README.rst

@@ -59,6 +59,22 @@ the following:
 * `PyAV <https://github.com/PyAV-Org/PyAV>`_
 
 
+Virus Scanning
+--------------
+
+0x0 can scan its files with ClamAV’s daemon. As this can take a long time
+for larger files, this does not happen immediately but instead every time
+you run the ``vscan`` command. It is recommended to configure a systemd
+timer or cronjob to do this periodically. Examples are included::
+
+    0x0-vscan.service
+    0x0-vscan.timer
+
+Remember to adjust your size limits in clamd.conf!
+
+This feature requires the `clamd module <https://pypi.org/project/clamd/>`_.
+
+
 Network Security Considerations
 -------------------------------
 

fhost.py

@@ -22,7 +22,7 @@
 from flask import Flask, abort, make_response, redirect, request, send_from_directory, url_for, Response, render_template
 from flask_sqlalchemy import SQLAlchemy
 from flask_migrate import Migrate
-from sqlalchemy import and_
+from sqlalchemy import and_, or_
 from jinja2.exceptions import *
 from jinja2 import ChoiceLoader, FileSystemLoader
 from hashlib import sha256
@@ -32,6 +32,7 @@ import click
 import os
 import sys
 import time
+import datetime
 import typing
 import requests
 import secrets
@@ -70,6 +71,13 @@ app.config.update(
     FHOST_UPLOAD_BLACKLIST = None,
     NSFW_DETECT = False,
     NSFW_THRESHOLD = 0.608,
+    VSCAN_SOCKET = None,
+    VSCAN_QUARANTINE_PATH = "quarantine",
+    VSCAN_IGNORE = [
+        "Eicar-Test-Signature",
+        "PUA.Win.Packer.XmMusicFile",
+    ],
+    VSCAN_INTERVAL = datetime.timedelta(days=7),
     URL_ALPHABET = "DEQhd2uFteibPwq0SWBInTpA_jcZL5GKz3YCR14Ulk87Jors9vNHgfaOmMXy6Vx-",
 )
 
@@ -131,6 +139,7 @@ class File(db.Model):
     expiration = db.Column(db.BigInteger)
     mgmt_token = db.Column(db.String)
     secret = db.Column(db.String)
+    last_vscan = db.Column(db.DateTime)
 
     def __init__(self, sha256, ext, mime, addr, expiration, mgmt_token):
         self.sha256 = sha256
@@ -591,3 +600,56 @@ def get_max_lifespan(filesize: int) -> int:
     max_exp = app.config.get("FHOST_MAX_EXPIRATION", 365 * 24 * 60 * 60 * 1000)
     max_size = app.config.get("MAX_CONTENT_LENGTH", 256 * 1024 * 1024)
     return min_exp + int((-max_exp + min_exp) * (filesize / max_size - 1) ** 3)
+
+def do_vscan(f):
+    if f["path"].is_file():
+        with open(f["path"], "rb") as scanf:
+            try:
+                f["result"] = list(app.config["VSCAN_SOCKET"].instream(scanf).values())[0]
+            except:
+                f["result"] = ("SCAN FAILED", None)
+    else:
+        f["result"] = ("FILE NOT FOUND", None)
+
+    return f
+
+@app.cli.command("vscan")
+def vscan():
+    if not app.config["VSCAN_SOCKET"]:
+        print("""Error: Virus scanning enabled but no connection method specified.
+Please set VSCAN_SOCKET.""")
+        sys.exit(1)
+
+    qp = Path(app.config["VSCAN_QUARANTINE_PATH"])
+    qp.mkdir(parents=True, exist_ok=True)
+
+    from multiprocessing import Pool
+    with Pool() as p:
+        if isinstance(app.config["VSCAN_INTERVAL"], datetime.timedelta):
+            scandate = datetime.datetime.now() - app.config["VSCAN_INTERVAL"]
+            res = File.query.filter(or_(File.last_vscan < scandate,
+                                        File.last_vscan == None),
+                                    File.removed == False)
+        else:
+            res = File.query.filter(File.last_vscan == None, File.removed == False)
+
+        work = [{"path" : f.getpath(), "name" : f.getname(), "id" : f.id} for f in res]
+
+        results = []
+        for i, r in enumerate(p.imap_unordered(do_vscan, work)):
+            if r["result"][0] != "OK":
+                print(f"{r['name']}: {r['result'][0]} {r['result'][1] or ''}")
+
+            found = False
+            if r["result"][0] == "FOUND":
+                if not r["result"][1] in app.config["VSCAN_IGNORE"]:
+                    r["path"].rename(qp / r["name"])
+                    found = True
+
+            results.append({
+                "id" : r["id"],
+                "last_vscan" : None if r["result"][0] == "SCAN FAILED" else datetime.datetime.now(),
+                "removed" : found})
+
+        db.session.bulk_update_mappings(File, results)
+        db.session.commit()

instance/config.example.py

@@ -168,6 +168,37 @@ NSFW_DETECT = False
 NSFW_THRESHOLD = 0.608
 
 
+# If you want to scan files for viruses using ClamAV, specify the socket used
+# for connections here. You will need the clamd module.
+# Since this can take a very long time on larger files, it is not done
+# immediately but every time you run the vscan command. It is recommended to
+# configure a systemd timer or cronjob to do this periodically.
+# Remember to adjust your size limits in clamd.conf!
+#
+# Example:
+# from clamd import ClamdUnixSocket
+# VSCAN_SOCKET = ClamdUnixSocket("/run/clamav/clamd-socket")
+
+# This is the directory that files flagged as malicious are moved to.
+# Relative paths are resolved relative to the working directory
+# of the 0x0 process.
+VSCAN_QUARANTINE_PATH = "quarantine"
+
+# Since updated virus definitions might catch some files that were previously
+# reported as clean, you may want to rescan old files periodically.
+# Set this to a datetime.timedelta to specify the frequency, or None to
+# disable rescanning.
+from datetime import timedelta
+VSCAN_INTERVAL = timedelta(days=7)
+
+# Some files flagged by ClamAV are usually not malicious, especially if the
+# DetectPUA option is enabled in clamd.conf. This is a list of signatures
+# that will be ignored.
+VSCAN_IGNORE = [
+    "Eicar-Test-Signature",
+    "PUA.Win.Packer.XmMusicFile",
+]
+
 # A list of all characters which can appear in a URL
 #
 # If this list is too short, then URLs can very quickly become long.

migrations/versions/5cee97aab219_.py

@@ -0,0 +1,26 @@
+"""add date of last virus scan
+
+Revision ID: 5cee97aab219
+Revises: e2e816056589
+Create Date: 2022-12-10 16:39:56.388259
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = '5cee97aab219'
+down_revision = 'e2e816056589'
+
+from alembic import op
+import sqlalchemy as sa
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('file', sa.Column('last_vscan', sa.DateTime(), nullable=True))
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('file', 'last_vscan')
+    # ### end Alembic commands ###