Harden OnJoin quote handling

Author: Alcheri

__init__.py

@@ -10,6 +10,7 @@
 """
 
 import sys
+from typing import Dict, List
 
 if sys.version_info <= (3, 6):
     raise RuntimeError("This plugin requires Python 3.6 or above.")
@@ -22,11 +23,13 @@
 __version__ = "1.1.0"
 
 # XXX Replace this with an appropriate author or supybot.Author instance.
-__author__ = supybot.Author("Barry Suridge", "Alcheri", "barry.suridge@outlook.com")
+__author__ = supybot.Author(
+    "Barry Suridge", "Alcheri", "barry.suridge@outlook.com"
+)
 
 # This is a dictionary mapping supybot.Author instances to lists of
 # contributions.
-__contributors__ = {}
+__contributors__: Dict[object, List[object]] = {}
 
 # This is a url where the most recent plugin package can be downloaded.
 __url__ = "https://github.com/Alcheri/Plugins.git"
@@ -42,7 +45,7 @@
 # reloaded when this plugin is reloaded.  Don't forget to import them as well!
 
 if world.testing:
-    from . import test
+    from . import test as test
 
 Class = plugin.Class
 configure = config.configure

config.py

@@ -12,13 +12,13 @@
     from supybot.i18n import PluginInternationalization
 
     _ = PluginInternationalization("OnJoin")
-except:
-    _ = lambda x: x
+except ImportError:
 
+    def _(text):
+        return text
 
-def configure(advanced):
-    from supybot.questions import expect, anything, something, yn
 
+def configure(advanced):
     conf.registerPlugin("OnJoin", True)
 
 
@@ -43,7 +43,9 @@ def configure(advanced):
 )
 
 conf.registerChannelValue(
-    OnJoin, "enable", registry.Boolean(False, """Should plugin work in this channel?""")
+    OnJoin,
+    "enable",
+    registry.Boolean(False, """Should plugin work in this channel?"""),
 )
 
 # vim:set shiftwidth=4 tabstop=4 expandtab textwidth=79:

plugin.py

@@ -3,45 +3,98 @@
 # All rights reserved.
 #
 
+import os
 import random
+import re
+import tempfile
+import threading
 from pathlib import Path
 
 import supybot.ircutils as utils
 import supybot.callbacks as callbacks
 from supybot.commands import optional, wrap
 
 DEFAULT_RECENT_QUOTES = 5
+HARD_MAX_QUOTES = 2000
+HARD_RECENT_QUOTES = 25
+MAX_QUOTE_LENGTH = 360
+CONTROL_CHARS_RE = re.compile(r"[\x00-\x1f\x7f]")
 
 
 class OnJoin(callbacks.Plugin):
     """Send a notice to all users entering a channel."""
 
     public = False
 
+    def __init__(self, irc):
+        super().__init__(irc)
+        self._quote_lock = threading.Lock()
+
     def _quotes_path(self):
         return Path(__file__).with_name("quotes.txt")
 
     def _normalise_quote(self, text):
-        quote = " ".join(text.splitlines()).strip()
+        quote = " ".join(str(text).splitlines()).strip()
+        quote = CONTROL_CHARS_RE.sub("", quote).strip()
+        if len(quote) > MAX_QUOTE_LENGTH:
+            quote = f"{quote[: MAX_QUOTE_LENGTH - 3]}..."
         return quote or None
 
+    def _max_stored_quotes(self):
+        try:
+            configured = self.registryValue("maxQuotes")
+        except Exception:
+            configured = HARD_MAX_QUOTES
+        return min(max(configured, 1), HARD_MAX_QUOTES)
+
+    def _quote_file_lock(self):
+        if not hasattr(self, "_quote_lock"):
+            self._quote_lock = threading.Lock()
+        return self._quote_lock
+
     def _load_quotes(self):
         quotes_path = self._quotes_path()
+        max_quotes = self._max_stored_quotes()
+        quotes = []
         try:
             with quotes_path.open(encoding="utf-8") as quote_file:
-                return [line.rstrip("\n") for line in quote_file if line.strip()]
+                for line in quote_file:
+                    quote = self._normalise_quote(line)
+                    if quote is None:
+                        continue
+                    quotes.append(quote)
+                    if len(quotes) > max_quotes:
+                        quotes.pop(0)
         except OSError as err:
             self.log.warning("OnJoin: failed to read %s: %s", quotes_path, err)
             return None
+        return quotes
 
     def _write_quotes(self, quotes):
         quotes_path = self._quotes_path()
+        tmp_name = None
         try:
-            with quotes_path.open("w", encoding="utf-8") as quote_file:
+            with tempfile.NamedTemporaryFile(
+                "w",
+                delete=False,
+                dir=str(quotes_path.parent),
+                encoding="utf-8",
+            ) as quote_file:
+                tmp_name = quote_file.name
                 for quote in quotes:
-                    quote_file.write(f"{quote}\n")
+                    normalised = self._normalise_quote(quote)
+                    if normalised:
+                        quote_file.write(f"{normalised}\n")
+            os.replace(tmp_name, str(quotes_path))
         except OSError as err:
-            self.log.warning("OnJoin: failed to write %s: %s", quotes_path, err)
+            self.log.warning(
+                "OnJoin: failed to write %s: %s", quotes_path, err
+            )
+            if tmp_name:
+                try:
+                    os.unlink(tmp_name)
+                except OSError:
+                    pass
             return False
         return True
 
@@ -50,38 +103,41 @@ def _append_quote(self, text):
         if quote is None:
             return None
 
-        quotes = self._load_quotes()
-        if quotes is None:
-            return False
+        with self._quote_file_lock():
+            quotes = self._load_quotes()
+            if quotes is None:
+                return False
 
-        quotes.append(quote)
-        max_quotes = self.registryValue("maxQuotes")
-        if len(quotes) > max_quotes:
-            quotes = quotes[-max_quotes:]
+            quotes.append(quote)
+            max_quotes = self._max_stored_quotes()
+            if len(quotes) > max_quotes:
+                quotes = quotes[-max_quotes:]
 
-        if not self._write_quotes(quotes):
-            return False
+            if not self._write_quotes(quotes):
+                return False
         return quote
 
     def _recent_quotes(self, count):
+        count = min(count, HARD_RECENT_QUOTES)
         quotes = self._load_quotes()
         if quotes is None:
             return None
         start_index = max(len(quotes) - count, 0) + 1
         return list(enumerate(quotes[-count:], start=start_index))
 
     def _delete_quote(self, quote_number):
-        quotes = self._load_quotes()
-        if quotes is None:
-            return False
-
-        index = quote_number - 1
-        if index < 0 or index >= len(quotes):
-            return None
-
-        deleted_quote = quotes.pop(index)
-        if not self._write_quotes(quotes):
-            return False
+        with self._quote_file_lock():
+            quotes = self._load_quotes()
+            if quotes is None:
+                return False
+
+            index = quote_number - 1
+            if index < 0 or index >= len(quotes):
+                return None
+
+            deleted_quote = quotes.pop(index)
+            if not self._write_quotes(quotes):
+                return False
         return deleted_quote
 
     def doJoin(self, irc, msg):
@@ -117,12 +173,12 @@ def _read_random_quote(self):
                     if not line.strip():
                         continue
                     line_num += 1
-                    if random.uniform(0, line_num) < 1:
+                    if random.uniform(0, line_num) < 1:  # nosec B311
                         selected_line = line
         except OSError as err:
             self.log.warning("OnJoin: failed to read %s: %s", quotes_path, err)
             return None
-        return selected_line or None
+        return self._normalise_quote(selected_line) if selected_line else None
 
     def addquote(self, irc, msg, args, text):
         """<text>
@@ -150,7 +206,7 @@ def recentquotes(self, irc, msg, args, count):
 
         max_count = self.registryValue("maxRecentQuotes")
         quote_count = count or DEFAULT_RECENT_QUOTES
-        quote_count = min(quote_count, max_count)
+        quote_count = min(quote_count, max_count, HARD_RECENT_QUOTES)
 
         quotes = self._recent_quotes(quote_count)
         if quotes is None:
@@ -161,7 +217,11 @@ def recentquotes(self, irc, msg, args, count):
             return
 
         for quote_number, quote in reversed(quotes):
-            irc.reply(f"{quote_number}. {quote}", notice=True, private=True)
+            irc.reply(
+                self._normalise_quote(f"{quote_number}. {quote}"),
+                notice=True,
+                private=True,
+            )
 
     recentquotes = wrap(
         recentquotes,

pyproject.toml

@@ -34,6 +34,15 @@ extend-exclude = '''
 )/
 '''
 
+[tool.pytest.ini_options]
+python_files = ["test.py"]
+
+[tool.mypy]
+python_version = "3.10"
+ignore_missing_imports = true
+explicit_package_bases = true
+exclude = "(__init__|test)\\.py"
+
 [tool.setuptools.package-dir]
 "limnoria_onjoin" = "."
 "limnoria_onjoin.local" = "local/"