Implement request filters

This moves preexisting blacklists to the database, and adds the
following filter types:

* IP address
* IP network
* MIME type
* User agent

In addition, IP address handling is now done with the ipaddress
module.

fhost.py

···
     and limitations under the License.
 """
 
-from flask import Flask, abort, make_response, redirect, request, send_from_directory, url_for, Response, render_template
+from flask import Flask, abort, make_response, redirect, request, send_from_directory, url_for, Response, render_template, Request
 from flask_sqlalchemy import SQLAlchemy
 from flask_migrate import Migrate
 from sqlalchemy import and_, or_
+from sqlalchemy.orm import declared_attr
+import sqlalchemy.types as types
 from jinja2.exceptions import *
 from jinja2 import ChoiceLoader, FileSystemLoader
 from hashlib import sha256
 from magic import Magic
 from mimetypes import guess_extension
 import click
+import enum
 import os
 import sys
 import time
 import datetime
+import ipaddress
 import typing
 import requests
 import secrets
+import re
 from validators import url as url_valid
 from pathlib import Path
 
···
         "text/plain" : ".txt",
         "text/x-diff" : ".diff",
     },
-    FHOST_MIME_BLACKLIST = [
-        "application/x-dosexec",
-        "application/java-archive",
-        "application/java-vm"
-    ],
-    FHOST_UPLOAD_BLACKLIST = None,
     NSFW_DETECT = False,
     NSFW_THRESHOLD = 0.92,
     VSCAN_SOCKET = None,
···
 
         return u
 
+class IPAddress(types.TypeDecorator):
+    impl = types.LargeBinary
+    cache_ok = True
+
+    def process_bind_param(self, value, dialect):
+        match value:
+            case ipaddress.IPv6Address():
+                value = (value.ipv4_mapped or value).packed
+            case ipaddress.IPv4Address():
+                value = value.packed
+
+        return value
+
+    def process_result_value(self, value, dialect):
+        if value is not None:
+            value = ipaddress.ip_address(value)
+            if type(value) is ipaddress.IPv6Address:
+                value = value.ipv4_mapped or value
+
+        return value
+
+
 class File(db.Model):
     id = db.Column(db.Integer, primary_key = True)
     sha256 = db.Column(db.String, unique = True)
     ext = db.Column(db.UnicodeText)
     mime = db.Column(db.UnicodeText)
-    addr = db.Column(db.UnicodeText)
+    addr = db.Column(IPAddress(16))
     ua = db.Column(db.UnicodeText)
     removed = db.Column(db.Boolean, default=False)
     nsfw_score = db.Column(db.Float)
···
             else:
                 mime = file_.content_type
 
-            if mime in app.config["FHOST_MIME_BLACKLIST"] or guess in app.config["FHOST_MIME_BLACKLIST"]:
-                abort(415)
-
             if len(mime) > 128:
                 abort(400)
+
+            for flt in MIMEFilter.query.all():
+                if flt.check(guess):
+                    abort(403, flt.reason)
 
             if mime.startswith("text/") and not "charset" in mime:
                 mime += "; charset=utf-8"
···
         return f, isnew
 
 
+class RequestFilter(db.Model):
+    __tablename__ = "request_filter"
+    id = db.Column(db.Integer, primary_key=True)
+    type = db.Column(db.String(20), index=True, nullable=False)
+    comment = db.Column(db.UnicodeText)
+
+    __mapper_args__ = {
+        "polymorphic_on": type,
+        "with_polymorphic": "*",
+        "polymorphic_identity": "empty"
+    }
+
+    def __init__(self, comment: str = None):
+        self.comment = comment
+
+
+class AddrFilter(RequestFilter):
+    addr = db.Column(IPAddress(16), unique=True)
+
+    __mapper_args__ = {"polymorphic_identity": "addr"}
+
+    def __init__(self, addr: ipaddress._BaseAddress, comment: str = None):
+        self.addr = addr
+        super().__init__(comment=comment)
+
+    def check(self, addr: ipaddress._BaseAddress) -> bool:
+        if type(addr) is ipaddress.IPv6Address:
+            addr = addr.ipv4_mapped or addr
+        return addr == self.addr
+
+    def check_request(self, r: Request) -> bool:
+        return self.check(ipaddress.ip_address(r.remote_addr))
+
+    @property
+    def reason(self) -> str:
+        return f"Your IP Address ({self.addr.compressed}) is blocked from " \
+                "uploading files."
+
+
+class IPNetwork(types.TypeDecorator):
+    impl = types.Text
+    cache_ok = True
+
+    def process_bind_param(self, value, dialect):
+        if value is not None:
+            value = value.compressed
+
+        return value
+
+    def process_result_value(self, value, dialect):
+        if value is not None:
+            value = ipaddress.ip_network(value)
+
+        return value
+
+
+class NetFilter(RequestFilter):
+    net = db.Column(IPNetwork)
+
+    __mapper_args__ = {"polymorphic_identity": "net"}
+
+    def __init__(self, net: ipaddress._BaseNetwork, comment: str = None):
+        self.net = net
+        super().__init__(comment=comment)
+
+    def check(self, addr: ipaddress._BaseAddress) -> bool:
+        if type(addr) is ipaddress.IPv6Address:
+            addr = addr.ipv4_mapped or addr
+        return addr in self.net
+
+    def check_request(self, r: Request) -> bool:
+        return self.check(ipaddress.ip_address(r.remote_addr))
+
+    @property
+    def reason(self) -> str:
+        return f"Your network ({self.net.compressed}) is blocked from " \
+                "uploading files."
+
+
+class HasRegex:
+    @declared_attr
+    def regex(cls):
+        return cls.__table__.c.get("regex", db.Column(db.UnicodeText))
+
+    def check(self, s: str) -> bool:
+        return re.match(self.regex, s) is not None
+
+
+class MIMEFilter(HasRegex, RequestFilter):
+    __mapper_args__ = {"polymorphic_identity": "mime"}
+
+    def __init__(self, mime_regex: str, comment: str = None):
+        self.regex = mime_regex
+        super().__init__(comment=comment)
+
+    def check_request(self, r: Request) -> bool:
+        if "file" in r.files:
+            return self.check(r.files["file"].mimetype)
+
+        return False
+
+    @property
+    def reason(self) -> str:
+        return "File MIME type not allowed."
+
+
+class UAFilter(HasRegex, RequestFilter):
+    __mapper_args__ = {"polymorphic_identity": "ua"}
+
+    def __init__(self, ua_regex: str, comment: str = None):
+        self.regex = ua_regex
+        super().__init__(comment=comment)
+
+    def check_request(self, r: Request) -> bool:
+        return self.check(r.user_agent.string)
+
+    @property
+    def reason(self) -> str:
+        return "User agent not allowed."
+
+
 class UrlEncoder(object):
     def __init__(self,alphabet, min_length):
         self.alphabet = alphabet
···
 
     return u.geturl()
 
-def in_upload_bl(addr):
-    if app.config["FHOST_UPLOAD_BLACKLIST"]:
-        with app.open_instance_resource(app.config["FHOST_UPLOAD_BLACKLIST"], "r") as bl:
-            check = addr.lstrip("::ffff:")
-            for l in bl.readlines():
-                if not l.startswith("#"):
-                    if check == l.rstrip():
-                        return True
-
-    return False
-
 """
 requested_expiration can be:
     - None, to use the longest allowed file lifespan
···
 Any value greater that the longest allowed file lifespan will be rounded down to that
 value.
 """
-def store_file(f, requested_expiration:  typing.Optional[int], addr, ua, secret: bool):
-    if in_upload_bl(addr):
-        return "Your host is blocked from uploading files.\n", 451
-
+def store_file(f, requested_expiration: typing.Optional[int], addr, ua, secret: bool):
     sf, isnew = File.store(f, requested_expiration, addr, ua, secret)
 
     response = make_response(sf.geturl())
···
 @app.route("/", methods=["GET", "POST"])
 def fhost():
     if request.method == "POST":
+        for flt in RequestFilter.query.all():
+            if flt.check_request(request):
+                abort(403, flt.reason)
+
         sf = None
         secret = "secret" in request.form
+        addr = ipaddress.ip_address(request.remote_addr)
+        if type(addr) is ipaddress.IPv6Address:
+            addr = addr.ipv4_mapped or addr
 
         if "file" in request.files:
             try:
···
                 return store_file(
                     request.files["file"],
                     int(request.form["expires"]),
-                    request.remote_addr,
+                    addr,
                     request.user_agent.string,
                     secret
                 )
···
                 return store_file(
                     request.files["file"],
                     None,
-                    request.remote_addr,
+                    addr,
                     request.user_agent.string,
                     secret
                 )
         elif "url" in request.form:
             return store_url(
                 request.form["url"],
-                request.remote_addr,
+                addr,
                 request.user_agent.string,
                 secret
             )
···
 
 @app.errorhandler(400)
 @app.errorhandler(401)
[e-mail](403)
 @app.errorhandler(404)
 @app.errorhandler(411)
 @app.errorhandler(413)
···
 @app.errorhandler(451)
 def ehandler(e):
     try:
-        return render_template(f"{e.code}.html", id=id, request=request), e.code
+        return render_template(f"{e.code}.html", id=id, request=request, description=e.description), e.code
     except TemplateNotFound:
         return "Segmentation fault\n", e.code
 

instance/config.example.py

···
     "text/x-diff" : ".diff",
 }
 
-
-# Control which files aren't allowed to be uploaded
-#
-# Certain kinds of files are never accepted.  If the file claims to be one of
-# these types of files, or if we look at the contents of the file and it looks
-# like one of these filetypes, then we reject the file outright with a 415
-# UNSUPPORTED MEDIA EXCEPTION
-FHOST_MIME_BLACKLIST = [
-    "application/x-dosexec",
-    "application/java-archive",
-    "application/java-vm"
-]
-
-
-# A list of IP addresses which are blacklisted from uploading files
-#
-# Can be set to the path of a file with an IP address on each line.  The file
-# can also include comment lines using a pound sign (#).  Paths are resolved
-# relative to the instance/ directory.
-#
-# If this is set to None, then no IP blacklist will be consulted.
-FHOST_UPLOAD_BLACKLIST = None
-
-
 # Enables support for detecting NSFW images
 #
 # Consult README.md for additional dependencies before setting to True

migrations/versions/5cda1743b92d_add_request_filters.py

···
+"""Add request filters
+
+Revision ID: 5cda1743b92d
+Revises: dd0766afb7d2
+Create Date: 2024-09-27 12:13:16.845981
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = '5cda1743b92d'
+down_revision = 'dd0766afb7d2'
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.ext.automap import automap_base
+from sqlalchemy.orm import Session
+from flask import current_app
+import ipaddress
+
+Base = automap_base()
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('request_filter',
+    sa.Column('id', sa.Integer(), nullable=False),
+    sa.Column('type', sa.String(length=20), nullable=False),
+    sa.Column('comment', sa.UnicodeText(), nullable=True),
+    sa.Column('addr', sa.LargeBinary(length=16), nullable=True),
+    sa.Column('net', sa.Text(), nullable=True),
+    sa.Column('regex', sa.UnicodeText(), nullable=True),
+    sa.PrimaryKeyConstraint('id'),
+    sa.UniqueConstraint('addr')
+    )
+    with op.batch_alter_table('request_filter', schema=None) as batch_op:
+        batch_op.create_index(batch_op.f('ix_request_filter_type'), ['type'], unique=False)
+
+    # ### end Alembic commands ###
+
+    bind = op.get_bind()
+    Base.prepare(autoload_with=bind)
+    RequestFilter = Base.classes.request_filter
+    session = Session(bind=bind)
+
+    if "FHOST_UPLOAD_BLACKLIST" in current_app.config:
+        if current_app.config["FHOST_UPLOAD_BLACKLIST"]:
+            with current_app.open_instance_resource(current_app.config["FHOST_UPLOAD_BLACKLIST"], "r") as bl:
+                for l in bl.readlines():
+                    if not l.startswith("#"):
+                        l = l.strip()
+                        if l.endswith(":"):
+                            # old implementation uses str.startswith,
+                            # which does not translate to networks
+                            current_app.logger.warning(f"Ignored address: {l}")
+                            continue
+
+                        flt = RequestFilter(type="addr", addr=ipaddress.ip_address(l).packed)
+                        session.add(flt)
+
+    if "FHOST_MIME_BLACKLIST" in current_app.config:
+        for mime in current_app.config["FHOST_MIME_BLACKLIST"]:
+            flt = RequestFilter(type="mime", regex=mime)
+            session.add(flt)
+
+    session.commit()
+
+    w = "Entries in your host and MIME blacklists have been migrated to " \
+        "request filters and stored in the databaes, where possible. " \
+        "The corresponding files and config options may now be deleted. " \
+        "Note that you may have to manually restore them if you wish to " \
+        "revert this with a db downgrade operation."
+    current_app.logger.warning(w)
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('request_filter', schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f('ix_request_filter_type'))
+
+    op.drop_table('request_filter')
+    # ### end Alembic commands ###

migrations/versions/d9a53a28ba54_change_file_addr_to_ipaddress_type.py

···
+"""Change File.addr to IPAddress type
+
+Revision ID: d9a53a28ba54
+Revises: 5cda1743b92d
+Create Date: 2024-09-27 14:03:06.764764
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = 'd9a53a28ba54'
+down_revision = '5cda1743b92d'
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.ext.automap import automap_base
+from sqlalchemy.orm import Session
+from flask import current_app
+import ipaddress
+
+Base = automap_base()
+
+
+def upgrade():
+    with op.batch_alter_table('file', schema=None) as batch_op:
+        batch_op.add_column(sa.Column('addr_tmp', sa.LargeBinary(16),
+                                      nullable=True))
+
+    bind = op.get_bind()
+    Base.prepare(autoload_with=bind)
+    File = Base.classes.file
+    session = Session(bind=bind)
+
+    updates = []
+    stmt = sa.select(File).where(sa.not_(File.addr == None))
+    for f in session.scalars(stmt.execution_options(yield_per=1000)):
+        addr = ipaddress.ip_address(f.addr)
+        if type(addr) is ipaddress.IPv6Address:
+            addr = addr.ipv4_mapped or addr
+
+        updates.append({
+            "id": f.id,
+            "addr_tmp": addr.packed
+        })
+    session.execute(sa.update(File), updates)
+
+    with op.batch_alter_table('file', schema=None) as batch_op:
+        batch_op.drop_column('addr')
+        batch_op.alter_column('addr_tmp', new_column_name='addr')
+
+
+def downgrade():
+    with op.batch_alter_table('file', schema=None) as batch_op:
+        batch_op.add_column(sa.Column('addr_tmp', sa.UnicodeText,
+                                      nullable=True))
+
+    bind = op.get_bind()
+    Base.prepare(autoload_with=bind)
+    File = Base.classes.file
+    session = Session(bind=bind)
+
+    updates = []
+    stmt = sa.select(File).where(sa.not_(File.addr == None))
+    for f in session.scalars(stmt.execution_options(yield_per=1000)):
+        addr = ipaddress.ip_address(f.addr)
+        if type(addr) is ipaddress.IPv6Address:
+            addr = addr.ipv4_mapped or addr
+
+        updates.append({
+            "id": f.id,
+            "addr_tmp": addr.compressed
+        })
+
+    session.execute(sa.update(File), updates)
+
+    with op.batch_alter_table('file', schema=None) as batch_op:
+        batch_op.drop_column('addr')
+        batch_op.alter_column('addr_tmp', new_column_name='addr')
+

mod.py

···
 from textual import log
 from rich.text import Text
 from jinja2.filters import do_filesizeformat
+import ipaddress
 
-from fhost import db, File, su, app as fhost_app, in_upload_bl
+from fhost import db, File, AddrFilter, su, app as fhost_app
 from modui import *
 
 fhost_app.app_context().push()
···
         if self.current_file:
             match fcol:
                 case 1: self.finput.value = ""
-                case 2: self.finput.value = self.current_file.addr
+                case 2: self.finput.value = self.current_file.addr.compressed
                 case 3: self.finput.value = self.current_file.mime
                 case 4: self.finput.value = self.current_file.ext
                 case 5: self.finput.value = self.current_file.ua or ""
···
                 case 1:
                     try: ftable.query = ftable.base_query.filter(File.id == su.debase(message.value))
                     except ValueError: pass
-                case 2: ftable.query = ftable.base_query.filter(File.addr.like(message.value))
+                case 2:
+                    try:
+                        addr = ipaddress.ip_address(message.value)
+                        if type(addr) is ipaddress.IPv6Address:
+                            addr = addr.ipv4_mapped or addr
+                        q = ftable.base_query.filter(File.addr == addr)
+                        ftable.query = q
+                    except ValueError: pass
                 case 3: ftable.query = ftable.base_query.filter(File.mime.like(message.value))
                 case 4: ftable.query = ftable.base_query.filter(File.ext.like(message.value))
                 case 5: ftable.query = ftable.base_query.filter(File.ua.like(message.value))
···
 
     def action_ban_ip(self, nuke: bool) -> None:
         if self.current_file:
-            if not fhost_app.config["FHOST_UPLOAD_BLACKLIST"]:
-                self.mount(Notification("Failed: FHOST_UPLOAD_BLACKLIST not set!"))
-                return
+            if AddrFilter.query.filter(AddrFilter.addr ==
+                                       self.current_file.addr).scalar():
+                txt = f"{self.current_file.addr.compressed} is already banned"
             else:
-                if in_upload_bl(self.current_file.addr):
-                    txt = f"{self.current_file.addr} is already banned"
-                else:
-                    with fhost_app.open_instance_resource(fhost_app.config["FHOST_UPLOAD_BLACKLIST"], "a") as bl:
-                        print(self.current_file.addr.lstrip("::ffff:"), file=bl)
-                    txt = f"Banned {self.current_file.addr}"
+                db.session.add(AddrFilter(self.current_file.addr))
+                db.session.commit()
+                txt = f"Banned {self.current_file.addr.compressed}"
 
-                if nuke:
-                    tsize = 0
-                    trm = 0
-                    for f in File.query.filter(File.addr == self.current_file.addr):
-                        if f.getpath().is_file():
-                            tsize += f.size or f.getpath().stat().st_size
-                            trm += 1
-                        f.delete(True)
-                    db.session.commit()
-                    txt += f", removed {trm} {'files' if trm != 1 else 'file'} totaling {do_filesizeformat(tsize, True)}"
+            if nuke:
+                tsize = 0
+                trm = 0
+                for f in File.query.filter(File.addr == self.current_file.addr):
+                    if f.getpath().is_file():
+                        tsize += f.size or f.getpath().stat().st_size
+                        trm += 1
+                    f.delete(True)
+                db.session.commit()
+                txt += f", removed {trm} {'files' if trm != 1 else 'file'} totaling {do_filesizeformat(tsize, True)}"
             self.mount(Notification(txt))
             self._refresh_layout()
             ftable = self.query_one("#ftable")
···
             ("File size:", do_filesizeformat(f.size, True)),
             ("MIME type:", f.mime),
             ("SHA256 checksum:", f.sha256),
-            ("Uploaded by:", Text(f.addr)),
+            ("Uploaded by:", Text(f.addr.compressed)),
             ("User agent:", Text(f.ua or "")),
             ("Management token:", f.mgmt_token),
             ("Secret:", f.secret),

requirements.txt

···
 Flask
 flask_sqlalchemy
 python_magic
+ipaddress
 
 # vscan
 clamd

templates/403.html

···
+{{ description if description else "Your host is banned." }}