new module mail_loop_prevention

This commit is contained in:
Marc Durepos 2025-10-07 10:06:10 -04:00
parent 18dc17a3bb
commit 7156cb440e
10 changed files with 750 additions and 0 deletions

View file

@ -0,0 +1,117 @@
# Mail Loop Prevention
## Overview
This module prevents communication loops that can occur when two mail servers auto-reply to each other indefinitely. This commonly happens with:
- Delivery receipt acknowledgements
- Out-of-office auto-replies
- Automated notification systems
- Email bounce handlers
## How It Works
The module overrides `mail.thread._message_create()` to detect potential loops by:
1. **Content Hashing**: Creates a hash of message body and subject (unescapes HTML and normalizes whitespace)
2. **Time Window Check**: Looks for identical messages within a configurable time window (default: 48 hours)
3. **Threshold Detection**: Blocks messages if the number of identical messages exceeds a threshold (default: 3)
4. **Smart Filtering**: Only checks automated messages (notification, auto_comment, email types), never blocks user comments
## Configuration
The module uses system parameters that can be configured via Settings > Technical > Parameters > System Parameters:
| Parameter | Default | Valid Range | Description |
|-----------|---------|-------------|-------------|
| `mail_loop_prevention.enabled` | `True` | `True`/`False` | Enable/disable loop prevention |
| `mail_loop_prevention.time_window_hours` | `48` | `1` to `720` | Time window in hours to check for duplicates (max 30 days) |
| `mail_loop_prevention.max_identical_messages` | `3` | `2` to `100` | Maximum identical messages allowed before blocking |
### Configuration Validation
The module validates all configuration parameters when they are set:
- **enabled**: Accepts `True`/`False`, `1`/`0`, `yes`/`no`, `on`/`off` (case-insensitive)
- **time_window_hours**: Must be an integer between 1 and 720 (30 days)
- **max_identical_messages**: Must be an integer between 2 and 100
Invalid values will raise a `ValidationError` and prevent the parameter from being saved.
## Example Scenario
**Without this module:**
```
Server A → Auto-reply to Server B
Server B → Auto-reply to Server A
Server A → Auto-reply to Server B
Server B → Auto-reply to Server A
... (infinite loop)
```
**With this module:**
```
Server A → Auto-reply to Server B (1st message - allowed)
Server B → Auto-reply to Server A (1st message - allowed)
Server A → Auto-reply to Server B (2nd message - allowed)
Server B → Auto-reply to Server A (2nd message - allowed)
Server A → Auto-reply to Server B (3rd message - allowed)
Server B → Auto-reply to Server A (3rd message - allowed)
Server A → Auto-reply to Server B (4th message - BLOCKED)
```
## Technical Details
### Message Types Checked
The module only checks these message types for loops:
- `notification` - System notifications
- `auto_comment` - Automated comments
- `email` - Email messages
Regular `comment` type messages (user posts) are never blocked.
### Duplicate Detection
Messages are considered identical if they have the same:
- Body content (after unescaping HTML entities and normalizing whitespace)
- Subject line
Real-world loops send **exactly the same message** repeatedly, so direct content comparison is sufficient.
### Performance Considerations
- Only checks messages within the configured time window
- Uses SHA-256 hashing for efficient comparison
- Minimal database queries (uses existing message_ids relation)
## Testing
Run the test suite:
```bash
odoo-bin -c odoo.conf -d test_db -i mail_loop_prevention --test-enable --stop-after-init
```
## Troubleshooting
### Legitimate messages being blocked
If legitimate automated messages are being blocked:
1. Check the logs for "Loop prevention: Blocking duplicate message" warnings
2. Increase `mail_loop_prevention.max_identical_messages` threshold
3. Reduce `mail_loop_prevention.time_window_hours` if the messages are spread over time
4. Temporarily disable with `mail_loop_prevention.enabled = False` to diagnose
### Loops still occurring
If loops are still happening:
1. Verify the module is installed and enabled
2. Check that `mail_loop_prevention.enabled = True`
3. Reduce the `max_identical_messages` threshold
4. Check if the messages have varying content (different subjects/bodies)
## License
AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)

View file

@ -0,0 +1 @@
from . import models

View file

@ -0,0 +1,17 @@
# Copyright 2025 DurPro
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
{
"name": "Mail Loop Prevention",
"summary": "Prevent auto-reply communication loops between mail servers",
"version": "17.0.1.0.0",
"license": "LGPL-3",
"author": "Bemade Inc",
"website": "https://github.com/durpro/durpro",
"depends": ["mail"],
"data": [
"views/res_config_settings_views.xml",
],
"installable": True,
"application": False,
}

View file

@ -0,0 +1,2 @@
from . import mail_thread
from . import res_config_settings

View file

@ -0,0 +1,162 @@
# Copyright 2025 DurPro
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
import hashlib
import logging
from datetime import timedelta
from odoo import api, models
from odoo.tools import html2plaintext
_logger = logging.getLogger(__name__)
class MailThread(models.AbstractModel):
_inherit = "mail.thread"
@api.model
def _get_loop_prevention_config(self):
"""
Get loop prevention configuration from system parameters.
Values are validated by ir.config_parameter model on write.
"""
ICP = self.env["ir.config_parameter"].sudo()
return {
"enabled": ICP.get_param(
"mail_loop_prevention.enabled", default="True"
) == "True",
"time_window_hours": int(
ICP.get_param("mail_loop_prevention.time_window_hours", default="48")
),
"max_identical_messages": int(
ICP.get_param("mail_loop_prevention.max_identical_messages", default="3")
),
}
def _compute_message_hash(self, body, subject=None):
"""
Compute a hash of the message content for duplicate detection.
Converts HTML to plaintext to ignore formatting differences.
"""
import html
# Unescape HTML entities (&lt; -> <, &gt; -> >, etc.)
# This is needed because message_post escapes the body before passing to _message_create
unescaped_body = html.unescape(body or "")
# Convert HTML to plaintext (handles extra wrapper tags that Odoo adds)
text_body = html2plaintext(unescaped_body).strip()
text_subject = (subject or "").strip()
# Combine subject and body for hash
content = f"{text_subject}|{text_body}"
# Debug logging to help troubleshoot loop detection
if _logger.isEnabledFor(logging.DEBUG):
_logger.debug(
"Loop prevention: Computing hash - plaintext=%r, hash=%s",
text_body[:100] if text_body else "",
hashlib.sha256(content.encode("utf-8")).hexdigest()[:16],
)
# Create hash
return hashlib.sha256(content.encode("utf-8")).hexdigest()
def _check_message_loop(self, body, subject=None, message_type="comment"):
"""
Check if posting this message would create a loop.
Returns True if the message should be blocked, False otherwise.
"""
config = self._get_loop_prevention_config()
if not config["enabled"]:
return False
# Only check for automated messages (notifications, auto_comment, etc.)
# Don't block regular user comments
if message_type not in ("notification", "auto_comment", "email"):
return False
# Compute hash of the new message
message_hash = self._compute_message_hash(body, subject)
# Calculate time threshold
time_threshold = self.env.cr.now() - timedelta(
hours=config["time_window_hours"]
)
# Search for identical messages in the time window
identical_count = 0
for record in self:
if not record.message_ids:
continue
identical_count = 0
for message in record.message_ids:
# Skip messages outside time window
if message.date < time_threshold:
continue
# Check if message content matches
msg_hash = self._compute_message_hash(message.body, message.subject)
if msg_hash == message_hash:
identical_count += 1
# If we've hit the limit, block the message
if identical_count >= config["max_identical_messages"]:
_logger.warning(
"Loop prevention: Blocking duplicate message on %s (id=%s). "
"Found %d identical messages in the last %d hours.",
record._name,
record.id,
identical_count,
config["time_window_hours"],
)
return True
return False
def _message_create(self, values_list):
"""
Override _message_create to check for message loops before creating messages.
This is safer than overriding message_post as we can cleanly filter out
messages that would create loops.
"""
filtered_values_list = []
for values in values_list:
body = values.get("body", "")
subject = values.get("subject", "")
message_type = values.get("message_type", "notification")
model = values.get("model")
res_id = values.get("res_id")
# Get the record to check its message history
if model and res_id:
try:
record = self.env[model].browse(res_id)
# All models with mail.thread mixin have _check_message_loop
if hasattr(record, '_check_message_loop'):
should_block = record._check_message_loop(body, subject, message_type)
if should_block:
_logger.warning(
"Loop prevention: Skipping message creation on %s (id=%s) "
"to prevent communication loop",
model,
res_id,
)
continue # Skip this message
except Exception as e:
# If we can't check, allow the message (fail open)
_logger.warning(
"Loop prevention: Error checking message loop, allowing message: %s",
e,
)
filtered_values_list.append(values)
# Create only the messages that passed the loop check
if not filtered_values_list:
return self.env["mail.message"]
return super()._message_create(filtered_values_list)

View file

@ -0,0 +1,55 @@
# Copyright 2025 DurPro
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
from odoo import api, fields, models
from odoo.exceptions import ValidationError
class ResConfigSettings(models.TransientModel):
_inherit = "res.config.settings"
mail_loop_prevention_enabled = fields.Boolean(
string="Enable Mail Loop Prevention",
config_parameter="mail_loop_prevention.enabled",
default=True,
help="Prevent auto-reply communication loops between mail servers",
)
mail_loop_prevention_time_window_hours = fields.Integer(
string="Time Window (hours)",
config_parameter="mail_loop_prevention.time_window_hours",
default=48,
help="Time window in hours to check for duplicate messages (1-720 hours)",
)
mail_loop_prevention_max_identical_messages = fields.Integer(
string="Maximum Identical Messages",
config_parameter="mail_loop_prevention.max_identical_messages",
default=3,
help="Maximum number of identical messages allowed before blocking (2-100)",
)
@api.constrains("mail_loop_prevention_time_window_hours")
def _check_time_window_hours(self):
for record in self:
if record.mail_loop_prevention_time_window_hours < 1:
raise ValidationError(
"Time window must be at least 1 hour."
)
if record.mail_loop_prevention_time_window_hours > 720:
raise ValidationError(
"Time window cannot exceed 720 hours (30 days)."
)
@api.constrains("mail_loop_prevention_max_identical_messages")
def _check_max_identical_messages(self):
for record in self:
if record.mail_loop_prevention_max_identical_messages < 2:
raise ValidationError(
"Maximum identical messages must be at least 2 "
"(setting to 1 would block the first duplicate)."
)
if record.mail_loop_prevention_max_identical_messages > 100:
raise ValidationError(
"Maximum identical messages cannot exceed 100."
)

View file

@ -0,0 +1,77 @@
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<title>Mail Loop Prevention</title>
</head>
<body>
<section class="oe_container">
<div class="oe_row oe_spaced">
<h2 class="oe_slogan">Mail Loop Prevention</h2>
<h3 class="oe_slogan">Stop Auto-Reply Communication Loops</h3>
<div class="oe_span12">
<p class="oe_mt32">
Prevent infinite communication loops when two mail servers auto-reply to each other.
This commonly occurs with delivery receipts, out-of-office messages, and automated notifications.
</p>
</div>
</div>
</section>
<section class="oe_container oe_dark">
<div class="oe_row oe_spaced">
<h2 class="oe_slogan">Features</h2>
<div class="oe_span6">
<div class="oe_demo oe_picture oe_screenshot">
<h3>Smart Detection</h3>
<p>
Detects duplicate messages by content hash with whitespace normalization.
Only checks automated messages - never blocks user comments.
</p>
</div>
</div>
<div class="oe_span6">
<div class="oe_demo oe_picture oe_screenshot">
<h3>Configurable</h3>
<p>
Customize time window (default 48 hours) and threshold (default 3 messages)
via system parameters.
</p>
</div>
</div>
</div>
</section>
<section class="oe_container">
<div class="oe_row oe_spaced">
<h2 class="oe_slogan">How It Works</h2>
<div class="oe_span12">
<ol>
<li>Monitors automated messages (notifications, emails, auto-comments)</li>
<li>Creates content hash of message body and subject</li>
<li>Checks for identical messages within time window</li>
<li>Blocks message if threshold is exceeded</li>
</ol>
<p class="oe_mt32">
<strong>Example:</strong> If 3 identical auto-replies are sent within 48 hours,
the 4th attempt will be blocked, preventing an infinite loop.
</p>
</div>
</div>
</section>
<section class="oe_container oe_dark">
<div class="oe_row oe_spaced">
<h2 class="oe_slogan">Configuration</h2>
<div class="oe_span12">
<p>Configure via Settings → Technical → Parameters → System Parameters:</p>
<ul>
<li><code>mail_loop_prevention.enabled</code> - Enable/disable (default: True)</li>
<li><code>mail_loop_prevention.time_window_hours</code> - Time window in hours (default: 48)</li>
<li><code>mail_loop_prevention.max_identical_messages</code> - Max duplicates allowed (default: 3)</li>
</ul>
</div>
</div>
</section>
</body>
</html>

View file

@ -0,0 +1 @@
from . import test_mail_loop_prevention

View file

@ -0,0 +1,289 @@
# Copyright 2025 DurPro
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
from datetime import timedelta
from freezegun import freeze_time
from odoo.exceptions import ValidationError
from odoo.tests import tagged
from odoo.tests.common import TransactionCase
@tagged("post_install", "-at_install")
class TestMailLoopPrevention(TransactionCase):
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.partner = cls.env["res.partner"].create(
{
"name": "Test Partner",
"email": "test@example.com",
}
)
cls.ICP = cls.env["ir.config_parameter"].sudo()
cls.Settings = cls.env["res.config.settings"]
def setUp(self):
super().setUp()
# Reset config to defaults for each test
self.ICP.set_param("mail_loop_prevention.enabled", "True")
self.ICP.set_param("mail_loop_prevention.time_window_hours", "48")
self.ICP.set_param("mail_loop_prevention.max_identical_messages", "3")
def test_normal_messages_not_blocked(self):
"""Test that normal user comments are never blocked."""
# Post the same message multiple times as a comment
for i in range(5):
msg = self.partner.message_post(
body="<p>This is a test message</p>",
subject="Test Subject",
message_type="comment",
)
self.assertTrue(msg, f"Message {i+1} should not be blocked")
def test_auto_reply_loop_detected(self):
"""Test that auto-reply loops are detected and blocked."""
auto_reply_body = "<p>Thank you for your message. We will respond shortly.</p>"
auto_reply_subject = "Auto-Reply: Message Received"
# Post the same auto-reply 3 times (should succeed)
for i in range(3):
msg = self.partner.message_post(
body=auto_reply_body,
subject=auto_reply_subject,
message_type="notification",
)
self.assertTrue(msg, f"Auto-reply {i+1} should be posted")
# The 4th identical message should be blocked
initial_count = len(self.partner.message_ids)
msg = self.partner.message_post(
body=auto_reply_body,
subject=auto_reply_subject,
message_type="notification",
)
# Message count should not increase
self.assertEqual(
len(self.partner.message_ids),
initial_count,
"4th identical message should be blocked",
)
def test_time_window_expiry(self):
"""Test that old messages outside time window don't count."""
auto_reply_body = "<p>Automated response</p>"
# Post 3 messages at different times
with freeze_time("2025-01-01 00:00:00"):
self.partner.message_post(
body=auto_reply_body,
message_type="notification",
)
with freeze_time("2025-01-02 00:00:00"):
self.partner.message_post(
body=auto_reply_body,
message_type="notification",
)
# 50 hours later (outside 48-hour window), the first message shouldn't count
with freeze_time("2025-01-03 02:00:00"):
# These should succeed because first message is outside window
for i in range(3):
msg = self.partner.message_post(
body=auto_reply_body,
message_type="notification",
)
self.assertTrue(msg, f"Message {i+1} should not be blocked")
def test_different_content_not_blocked(self):
"""Test that different messages are not blocked."""
# Post 5 different auto-replies
for i in range(5):
msg = self.partner.message_post(
body=f"<p>Auto-reply message {i}</p>",
message_type="notification",
)
self.assertTrue(msg, f"Different message {i+1} should not be blocked")
def test_identical_messages_blocked(self):
"""Test that identical messages are blocked."""
# Post the exact same message multiple times
message_body = "<p>Thank you for your message</p>"
# Post first 3 (should succeed)
for i in range(3):
msg = self.partner.message_post(
body=message_body,
message_type="notification",
)
self.assertTrue(msg, f"Message {i+1} should be posted")
# 4th identical message should be blocked
initial_count = len(self.partner.message_ids)
self.partner.message_post(
body=message_body,
message_type="notification",
)
self.assertEqual(
len(self.partner.message_ids),
initial_count,
"4th identical message should be blocked",
)
def test_config_disabled(self):
"""Test that loop prevention can be disabled."""
self.ICP.set_param("mail_loop_prevention.enabled", "False")
auto_reply_body = "<p>Auto-reply</p>"
# Post 5 identical messages - none should be blocked when disabled
for i in range(5):
msg = self.partner.message_post(
body=auto_reply_body,
message_type="notification",
)
self.assertTrue(msg, f"Message {i+1} should not be blocked when disabled")
def test_custom_threshold(self):
"""Test that custom threshold is respected."""
# Set threshold to 2 instead of 3
self.ICP.set_param("mail_loop_prevention.max_identical_messages", "2")
auto_reply_body = "<p>Auto-reply</p>"
# Post 2 messages (should succeed)
for i in range(2):
msg = self.partner.message_post(
body=auto_reply_body,
message_type="notification",
)
self.assertTrue(msg, f"Message {i+1} should be posted")
# 3rd should be blocked with threshold of 2
initial_count = len(self.partner.message_ids)
self.partner.message_post(
body=auto_reply_body,
message_type="notification",
)
self.assertEqual(
len(self.partner.message_ids),
initial_count,
"3rd message should be blocked with threshold of 2",
)
def test_email_message_type_checked(self):
"""Test that email message type is also checked for loops."""
email_body = "<p>Delivery notification</p>"
# Post 3 identical emails
for i in range(3):
msg = self.partner.message_post(
body=email_body,
message_type="email",
)
self.assertTrue(msg, f"Email {i+1} should be posted")
# 4th should be blocked
initial_count = len(self.partner.message_ids)
self.partner.message_post(
body=email_body,
message_type="email",
)
self.assertEqual(
len(self.partner.message_ids),
initial_count,
"4th identical email should be blocked",
)
def test_config_validation_invalid_time_window(self):
"""Test that invalid time_window_hours values are rejected."""
# Test too low
with self.assertRaises(ValidationError):
settings = self.Settings.create({
"mail_loop_prevention_time_window_hours": 0,
})
settings.execute()
with self.assertRaises(ValidationError):
settings = self.Settings.create({
"mail_loop_prevention_time_window_hours": -5,
})
settings.execute()
# Test too high
with self.assertRaises(ValidationError):
settings = self.Settings.create({
"mail_loop_prevention_time_window_hours": 1000,
})
settings.execute()
def test_config_validation_invalid_max_messages(self):
"""Test that invalid max_identical_messages values are rejected."""
# Test too low
with self.assertRaises(ValidationError):
settings = self.Settings.create({
"mail_loop_prevention_max_identical_messages": 1,
})
settings.execute()
with self.assertRaises(ValidationError):
settings = self.Settings.create({
"mail_loop_prevention_max_identical_messages": 0,
})
settings.execute()
# Test too high
with self.assertRaises(ValidationError):
settings = self.Settings.create({
"mail_loop_prevention_max_identical_messages": 200,
})
settings.execute()
def test_config_validation_valid_boundary_values(self):
"""Test that valid boundary values are accepted."""
# Test minimum valid values
settings = self.Settings.create({
"mail_loop_prevention_time_window_hours": 1,
"mail_loop_prevention_max_identical_messages": 2,
})
settings.execute()
self.assertEqual(
self.ICP.get_param("mail_loop_prevention.time_window_hours"), "1"
)
self.assertEqual(
self.ICP.get_param("mail_loop_prevention.max_identical_messages"), "2"
)
# Test maximum valid values
settings = self.Settings.create({
"mail_loop_prevention_time_window_hours": 720,
"mail_loop_prevention_max_identical_messages": 100,
})
settings.execute()
self.assertEqual(
self.ICP.get_param("mail_loop_prevention.time_window_hours"), "720"
)
self.assertEqual(
self.ICP.get_param("mail_loop_prevention.max_identical_messages"), "100"
)
def test_config_validation_enabled_param(self):
"""Test that enabled parameter can be toggled."""
# Test enabling
settings = self.Settings.create({
"mail_loop_prevention_enabled": True,
})
settings.execute()
self.assertEqual(
self.ICP.get_param("mail_loop_prevention.enabled"), "True"
)
# Test disabling
settings = self.Settings.create({
"mail_loop_prevention_enabled": False,
})
settings.execute()
self.assertEqual(
self.ICP.get_param("mail_loop_prevention.enabled"), "False"
)

View file

@ -0,0 +1,29 @@
<?xml version="1.0" encoding="utf-8"?>
<odoo>
<record id="res_config_settings_view_form" model="ir.ui.view">
<field name="name">res.config.settings.view.form.inherit.mail.loop.prevention</field>
<field name="model">res.config.settings</field>
<field name="inherit_id" ref="base_setup.res_config_settings_view_form"/>
<field name="arch" type="xml">
<xpath expr="//app[@name='general_settings']" position="inside">
<block title="Mail Loop Prevention">
<setting help="Prevent infinite communication loops when two mail servers auto-reply to each other">
<field name="mail_loop_prevention_enabled"/>
<div class="content-group" invisible="not mail_loop_prevention_enabled">
<div class="row mt16">
<label for="mail_loop_prevention_time_window_hours" string="Time Window" class="col-3 o_light_label"/>
<field name="mail_loop_prevention_time_window_hours"/>
<span>hours (1-720)</span>
</div>
<div class="row">
<label for="mail_loop_prevention_max_identical_messages" string="Maximum Duplicates" class="col-3 o_light_label"/>
<field name="mail_loop_prevention_max_identical_messages"/>
<span>messages (2-100)</span>
</div>
</div>
</setting>
</block>
</xpath>
</field>
</record>
</odoo>