β → 🜚🏭 ❻

KerbalStuff/antivirus.py

@@ -0,0 +1,48 @@
+import os
+from pathlib import Path
+from shutil import move
+from typing import Optional
+
+import pyclamd
+
+from .config import _cfg, _cfgi, site_logger
+from .objects import User
+from .email import send_mod_locked
+from .ckan import notify_ckan
+
+clam_daemon = None
+
+
+def file_contains_malware(where: str) -> bool:
+    global clam_daemon
+    try:
+        if not clam_daemon:
+            clam_daemon = pyclamd.ClamdNetworkSocket(host=_cfg('clamav-host'), port=_cfgi('clamav-port', 3310))
+        result = clam_daemon.scan_file(where)
+        if result:
+            site_logger.error(f'ClamAV says {where} contains malware')
+            return True
+    except Exception as exc:
+        # No ClamAV daemon found, log it and let the file through
+        site_logger.error(f'Failed to connect to ClamAV, skipping scan of {where}', exc_info=exc)
+    return False
+
+
+def quarantine_malware(path: str) -> None:
+    quarantine_folder = _cfg('clamav-quarantine-path')
+    if quarantine_folder:
+        move(path, Path(quarantine_folder) / Path(path).name)
+    else:
+        os.remove(path)
+
+
+def punish_malware(user: User) -> None:
+    # Lock all of this author's mods
+    for other_mod in user.mods:
+        if not other_mod.locked:
+            other_mod.locked = True
+            other_mod.published = False
+            other_mod.locked_by = None
+            other_mod.lock_reason = 'Malware detected in upload'
+            send_mod_locked(other_mod, user)
+            notify_ckan(other_mod, 'locked', True)

KerbalStuff/app.py

@@ -6,6 +6,7 @@
 import subprocess
 import xml.etree.ElementTree as ET
 from time import strftime
+from datetime import timedelta
 from typing import Tuple, List, Dict, Any, Optional, Union
 from pathlib import Path
 
@@ -14,16 +15,16 @@
 from flask import Flask, render_template, g, url_for, Response, request
 from flask_login import LoginManager, current_user
 from flaskext.markdown import Markdown
-from sqlalchemy import desc
 from werkzeug.exceptions import HTTPException, InternalServerError, NotFound
 from flask.typing import ResponseReturnValue
 from jinja2 import ChainableUndefined
+from pymdownx.emoji import gemoji, to_alt
 
 from .blueprints.accounts import accounts
 from .blueprints.admin import admin
 from .blueprints.anonymous import anonymous
 from .blueprints.api import api
-from .blueprints.blog import blog
+from .blueprints.blog import blog, get_all_announcement_posts, get_non_member_announcement_posts
 from .blueprints.lists import lists
 from .blueprints.login_oauth import list_defined_oauths, login_oauth
 from .blueprints.mods import mods
@@ -60,16 +61,29 @@
 app.secret_key = _cfg("secret-key")
 app.json_encoder = CustomJSONEncoder
 app.session_interface = OnlyLoggedInSessionInterface()
-Markdown(app, extensions=[KerbDown(), 'fenced_code'])
+Markdown(app, extensions=[KerbDown(), 'fenced_code', 'pymdownx.emoji'],
+         extension_configs={'pymdownx.emoji': {
+            # GitHub's emojis
+            'emoji_index': gemoji,
+            # Unicode output
+            'emoji_generator': to_alt
+         }})
 login_manager = LoginManager(app)
 
 prof_dir = _cfg('profile-dir')
 if prof_dir:
-    from .middleware.profiler import ConditionalProfilerMiddleware
-    from .profiling import sampling_function
-    Path(prof_dir).mkdir(parents=True, exist_ok=True)
-    app.wsgi_app = ConditionalProfilerMiddleware(  # type: ignore[assignment]
-        app.wsgi_app, stream=None, profile_dir=prof_dir, sampling_function=sampling_function)
+    log_if_longer = _cfg('profile-threshold-ms')
+    if log_if_longer:
+        from .middleware.profiler import CherrypickingProfilerMiddleware
+        Path(prof_dir).mkdir(parents=True, exist_ok=True)
+        app.wsgi_app = CherrypickingProfilerMiddleware(  # type: ignore[assignment]
+            app.wsgi_app, stream=None, profile_dir=prof_dir, log_if_longer=timedelta(milliseconds=int(log_if_longer)))
+    else:
+        from .middleware.profiler import ConditionalProfilerMiddleware
+        from .profiling import sampling_function
+        Path(prof_dir).mkdir(parents=True, exist_ok=True)
+        app.wsgi_app = ConditionalProfilerMiddleware(  # type: ignore[assignment]
+            app.wsgi_app, stream=None, profile_dir=prof_dir, sampling_function=sampling_function)
 
 
 @login_manager.user_loader
@@ -321,13 +335,3 @@ def inject() -> Dict[str, Any]:
         'donation_header_link': _cfgb('donation-header-link') if not dismissed_donation else False,
         'registration': _cfgb('registration')
     }
-
-
-def get_all_announcement_posts() -> List[BlogPost]:
-    return BlogPost.query.filter(BlogPost.announcement).order_by(desc(BlogPost.created)).all()
-
-
-def get_non_member_announcement_posts() -> List[BlogPost]:
-    return BlogPost.query.filter(
-        BlogPost.announcement, BlogPost.members_only != True
-    ).order_by(desc(BlogPost.created)).all()

KerbalStuff/blueprints/admin.py

@@ -1,13 +1,13 @@
 import math
-from typing import Union, List, Tuple, Dict, Any
+from typing import Union, List, Tuple, Dict, Any, Optional
 import datetime
 from datetime import timezone
 from pathlib import Path
 from subprocess import run, PIPE
 
 from flask import Blueprint, render_template, redirect, request, abort, url_for
 from flask_login import login_user, current_user
-from sqlalchemy import desc, or_, func
+from sqlalchemy import or_, func
 from sqlalchemy.orm import Query
 import werkzeug.wrappers
 
@@ -19,6 +19,7 @@
 
 admin = Blueprint('admin', __name__)
 ITEMS_PER_PAGE = 10
+MODS_PER_PAGE = 30
 
 
 @admin.route("/admin")
@@ -121,7 +122,7 @@ def users(page: int) -> Union[str, werkzeug.wrappers.Response]:
     users = search_users(query.lower()) if query else User.query
     if not show_non_public:
         users = users.filter(User.public)
-    users = users.order_by(desc(User.created))
+    users = users.order_by(User.created.desc())
     user_count = users.count()
     # We can limit here because SqlAlchemy executes queries lazily.
     users = users.offset((page - 1) * ITEMS_PER_PAGE).limit(ITEMS_PER_PAGE)
@@ -135,6 +136,23 @@ def users(page: int) -> Union[str, werkzeug.wrappers.Response]:
                            query=query, show_non_public=show_non_public)
 
 
+@admin.route("/admin/locked_mods/<int:page>")
+@adminrequired
+def locked_mods(page: int) -> Union[str, werkzeug.wrappers.Response]:
+    if page < 1:
+        return redirect(url_for('admin.locked_mods', page=1, **request.args))
+    locked_mods = Mod.query.filter(Mod.locked == True)\
+                           .order_by(Mod.updated.desc())
+    locked_mods_count = locked_mods.count()
+    total_pages = max(1, math.ceil(locked_mods_count / MODS_PER_PAGE))
+    if page > total_pages:
+        return redirect(url_for('admin.locked_mods', page=total_pages, **request.args))
+    return render_template("admin-locked-mods.html",
+                           page=page, total_pages=total_pages,
+                           locked_mods=locked_mods.offset((page - 1) * MODS_PER_PAGE)\
+                                                  .limit(MODS_PER_PAGE))
+
+
 @admin.route("/admin/blog")
 @adminrequired
 def blog() -> str:
@@ -143,15 +161,15 @@ def blog() -> str:
 
 @admin.route("/admin/publishers/<int:page>")
 @adminrequired
-def publishers(page: int, error: str = None) -> Union[str, werkzeug.wrappers.Response]:
+def publishers(page: int, error: Optional[str] = None) -> Union[str, werkzeug.wrappers.Response]:
     if page < 1:
         return redirect(url_for('admin.publishers', page=1, **request.args))
     show_none_active = (request.args.get('show_none_active', '').lower() in TRUE_STR)
     query = request.args.get('query', type=str)
     publishers = search_publishers(query.lower()) if query else Publisher.query
     if not show_none_active:
         publishers = publishers.join(Publisher.games).filter(Game.active).distinct(Publisher.id)
-    publishers = publishers.order_by(desc(Publisher.id))
+    publishers = publishers.order_by(Publisher.id.desc())
     publisher_count = publishers.count()
     publishers = publishers.offset((page - 1) * ITEMS_PER_PAGE).limit(ITEMS_PER_PAGE)
 
@@ -167,23 +185,23 @@ def publishers(page: int, error: str = None) -> Union[str, werkzeug.wrappers.Res
 
 @admin.route("/admin/games/<int:page>")
 @adminrequired
-def games(page: int, error: str = None) -> Union[str, werkzeug.wrappers.Response]:
+def games(page: int, error: Optional[str] = None) -> Union[str, werkzeug.wrappers.Response]:
     if page < 1:
         return redirect(url_for('admin.games', page=1, **request.args))
     show_inactive = (request.args.get('show_inactive', '').lower() in TRUE_STR)
     query = request.args.get('query', type=str)
     games = search_games(query.lower()) if query else Game.query
     if not show_inactive:
         games = games.filter(Game.active)
-    games = games.order_by(desc(Game.id))
+    games = games.order_by(Game.id.desc())
     game_count = games.count()
     games = games.offset((page - 1) * ITEMS_PER_PAGE).limit(ITEMS_PER_PAGE)
 
     total_pages = max(1, math.ceil(game_count / ITEMS_PER_PAGE))
     if page > total_pages:
         return redirect(url_for('admin.games', page=total_pages, **request.args))
 
-    publishers = Publisher.query.order_by(desc(Publisher.id))
+    publishers = Publisher.query.order_by(Publisher.id.desc())
 
     return render_template('admin-games.html',
                            games=games, publishers=publishers, game_count=game_count,
@@ -194,23 +212,23 @@ def games(page: int, error: str = None) -> Union[str, werkzeug.wrappers.Response
 
 @admin.route("/admin/gameversions/<int:page>")
 @adminrequired
-def game_versions(page: int, error: str = None) -> Union[str, werkzeug.wrappers.Response]:
+def game_versions(page: int, error: Optional[str] = None) -> Union[str, werkzeug.wrappers.Response]:
     if page < 1:
         return redirect(url_for('admin.game_versions', page=1, **request.args))
     show_inactive = (request.args.get('show_inactive', '').lower() in TRUE_STR)
     query = request.args.get('query', type=str)
     game_versions = search_game_versions(query.lower()) if query else GameVersion.query
     if not show_inactive:
         game_versions = game_versions.join(GameVersion.game).filter(Game.active)
-    game_versions = game_versions.order_by(desc(GameVersion.id))
+    game_versions = game_versions.order_by(GameVersion.id.desc())
     game_version_count = game_versions.count()
     game_versions = game_versions.offset((page - 1) * ITEMS_PER_PAGE).limit(ITEMS_PER_PAGE)
 
     total_pages = max(1, math.ceil(game_version_count / ITEMS_PER_PAGE))
     if page > total_pages:
         return redirect(url_for('admin.game_versions', page=total_pages, **request.args))
 
-    games = Game.query.order_by(desc(Game.id))
+    games = Game.query.order_by(Game.id.desc())
 
     return render_template('admin-game-versions.html',
                            game_versions=game_versions, games=games,

KerbalStuff/blueprints/anonymous.py

@@ -3,7 +3,6 @@
 import werkzeug.wrappers
 from flask import Blueprint, render_template, abort, request, Response, make_response, send_file
 from flask_login import current_user
-from sqlalchemy import desc
 from datetime import timezone
 
 from ..common import dumb_object, paginate_query, get_paginated_mods, get_game_info, get_games, \
@@ -61,7 +60,7 @@ def browse() -> str:
 
 @anonymous.route("/browse/new")
 def browse_new() -> str:
-    mods = Mod.query.filter(Mod.published).order_by(desc(Mod.created))
+    mods = Mod.query.filter(Mod.published).order_by(Mod.created.desc())
     mods, page, total_pages = paginate_query(mods)
     return render_template("browse-list.html", mods=mods, page=page, total_pages=total_pages,
                            url="/browse/new", name="Newest Mods", rss="/browse/new.rss")
@@ -80,7 +79,7 @@ def browse_new_rss() -> Response:
 
 @anonymous.route("/browse/updated")
 def browse_updated() -> str:
-    mods = Mod.query.filter(Mod.published, Mod.versions.any(ModVersion.id != Mod.default_version_id)).order_by(desc(Mod.updated))
+    mods = Mod.query.filter(Mod.published, Mod.versions.any(ModVersion.id != Mod.default_version_id)).order_by(Mod.updated.desc())
     mods, page, total_pages = paginate_query(mods)
     return render_template("browse-list.html", mods=mods, page=page, total_pages=total_pages,
                            url="/browse/updated", name="Recently Updated Mods", rss="/browse/updated.rss")
@@ -107,7 +106,7 @@ def browse_top() -> str:
 
 @anonymous.route("/browse/featured")
 def browse_featured() -> str:
-    mods = Featured.query.order_by(desc(Featured.created))
+    mods = Featured.query.order_by(Featured.created.desc())
     mods, page, total_pages = paginate_query(mods)
     mods = [f.mod for f in mods]
     return render_template("browse-list.html", mods=mods, page=page, total_pages=total_pages,
@@ -150,7 +149,7 @@ def singlegame_browse(gameshort: str) -> str:
 @anonymous.route("/<gameshort>/browse/new")
 def singlegame_browse_new(gameshort: str) -> str:
     ga = get_game_info(short=gameshort)
-    mods = Mod.query.filter(Mod.published, Mod.game_id == ga.id).order_by(desc(Mod.created))
+    mods = Mod.query.filter(Mod.published, Mod.game_id == ga.id).order_by(Mod.created.desc())
     mods, page, total_pages = paginate_query(mods)
     return render_template("browse-list.html", mods=mods, page=page, total_pages=total_pages, ga=ga,
                            url="/browse/new", name="Newest Mods", rss="/browse/new.rss")
@@ -172,7 +171,7 @@ def singlegame_browse_new_rss(gameshort: str) -> Response:
 @anonymous.route("/<gameshort>/browse/updated")
 def singlegame_browse_updated(gameshort: str) -> str:
     ga = get_game_info(short=gameshort)
-    mods = Mod.query.filter(Mod.published, Mod.game_id == ga.id, Mod.versions.any(ModVersion.id != Mod.default_version_id)).order_by(desc(Mod.updated))
+    mods = Mod.query.filter(Mod.published, Mod.game_id == ga.id, Mod.versions.any(ModVersion.id != Mod.default_version_id)).order_by(Mod.updated.desc())
     mods, page, total_pages = paginate_query(mods)
     return render_template("browse-list.html", mods=mods, page=page, total_pages=total_pages, ga=ga,
                            url="/browse/updated", name="Recently Updated Mods", rss="/browse/updated.rss")
@@ -204,7 +203,7 @@ def singlegame_browse_top(gameshort: str) -> str:
 def singlegame_browse_featured(gameshort: str) -> str:
     ga = get_game_info(short=gameshort)
     mods = Featured.query.outerjoin(Mod).filter(
-        Mod.game_id == ga.id).order_by(desc(Featured.created))
+        Mod.game_id == ga.id).order_by(Featured.created.desc())
     mods, page, total_pages = paginate_query(mods)
     mods = [f.mod for f in mods]
     return render_template("browse-list.html", mods=mods, page=page, total_pages=total_pages, ga=ga,