Escape revision messages before embedding in script.py.mako docstrings

alembic/script/base.py

@@ -46,6 +46,18 @@
 _default_file_template = "%(rev)s_%(slug)s"
 
 
+
+def _escape_docstring_message(message: str) -> str:
+    """Escape a revision message for safe embedding in a triple-quoted docstring.
+
+    Unescaped triple quotes would terminate the docstring and allow the rest of
+    the message to execute as module-level code on import. Backslashes must also
+    be doubled so the original message round-trips through ``script.doc``.
+    See https://github.com/sqlalchemy/alembic/issues/1879.
+    """
+    return message.replace("\\", "\\\\").replace('"""', '\\"\\"\\"')
+
+
 class ScriptDirectory:
     """Provides operations upon an Alembic script directory.
 
@@ -740,7 +752,9 @@ def generate_revision(
             depends_on=revision.tuple_rev_as_scalar(resolved_depends_on),
             create_date=create_date,
             comma=util.format_as_comma,
-            message=message if message is not None else ("empty message"),
+            message=_escape_docstring_message(message)
+            if message is not None
+            else "empty message",
             **kw,
         )
 

tests/test_script_production.py

@@ -406,6 +406,27 @@ def test_create_script_basic(self):
         eq_(rev.down_revision, self.c)
         assert "some message" in rev.doc
 
+    def test_revision_message_escapes_triple_quotes(self):
+        # Message must not terminate the generated module docstring (#1879).
+        message = 'add users"""\nprint("injected")'
+        message = (
+            "add users"
+            + '"""'
+            + chr(10)
+            + 'print("injected")'
+        )
+        rev = command.revision(self.cfg, message=message)
+        script = ScriptDirectory.from_config(self.cfg)
+        rev = script.get_revision(rev.revision)
+        eq_(rev.doc, message)
+
+    def test_revision_message_escapes_backslashes(self):
+        message = "C:" + chr(92) + "path" + chr(92) + "new"
+        rev = command.revision(self.cfg, message=message)
+        script = ScriptDirectory.from_config(self.cfg)
+        rev = script.get_revision(rev.revision)
+        eq_(rev.doc, message)
+
     def test_create_script_splice(self):
         rev = command.revision(
             self.cfg, message="some message", head=self.b, splice=True