policies/reputation: limit reputation score (#14008)

* add limits to reputation score

* limit reputation score limits

Upper to non-negative, Lower to non-positive

* simplify tests

* "fix" bandit false-positives

* move magic numbers to constants

Is it too much to ask for a world in which I can just import these
straight from Python?

Author: gergosimonyi

authentik/policies/reputation/signals.py

@@ -2,7 +2,6 @@
 
 from django.contrib.auth.signals import user_logged_in
 from django.db import transaction
-from django.db.models import F
 from django.dispatch import receiver
 from django.http import HttpRequest
 from structlog.stdlib import get_logger
@@ -13,30 +12,45 @@
 from authentik.policies.reputation.models import Reputation, reputation_expiry
 from authentik.root.middleware import ClientIPMiddleware
 from authentik.stages.identification.signals import identification_failed
+from authentik.tenants.utils import get_current_tenant
 
 LOGGER = get_logger()
 
 
+def clamp(value, min, max):
+    return sorted([min, value, max])[1]
+
+
 def update_score(request: HttpRequest, identifier: str, amount: int):
     """Update score for IP and User"""
     remote_ip = ClientIPMiddleware.get_client_ip(request)
+    tenant = get_current_tenant()
+    new_score = clamp(amount, tenant.reputation_lower_limit, tenant.reputation_upper_limit)
 
     with transaction.atomic():
         reputation, created = Reputation.objects.select_for_update().get_or_create(
             ip=remote_ip,
             identifier=identifier,
             defaults={
-                "score": amount,
+                "score": clamp(
+                    amount, tenant.reputation_lower_limit, tenant.reputation_upper_limit
+                ),
                 "ip_geo_data": GEOIP_CONTEXT_PROCESSOR.city_dict(remote_ip) or {},
                 "ip_asn_data": ASN_CONTEXT_PROCESSOR.asn_dict(remote_ip) or {},
                 "expires": reputation_expiry(),
             },
         )
 
         if not created:
-            reputation.score = F("score") + amount
+            new_score = clamp(
+                reputation.score + amount,
+                tenant.reputation_lower_limit,
+                tenant.reputation_upper_limit,
+            )
+            reputation.score = new_score
             reputation.save()
-    LOGGER.info("Updated score", amount=amount, for_user=identifier, for_ip=remote_ip)
+
+    LOGGER.info("Updated score", amount=new_score, for_user=identifier, for_ip=remote_ip)
 
 
 @receiver(login_failed)

authentik/policies/reputation/tests.py

@@ -6,9 +6,11 @@
 from authentik.lib.generators import generate_id
 from authentik.policies.reputation.api import ReputationPolicySerializer
 from authentik.policies.reputation.models import Reputation, ReputationPolicy
+from authentik.policies.reputation.signals import update_score
 from authentik.policies.types import PolicyRequest
 from authentik.stages.password import BACKEND_INBUILT
 from authentik.stages.password.stage import authenticate
+from authentik.tenants.models import DEFAULT_REPUTATION_LOWER_LIMIT, DEFAULT_REPUTATION_UPPER_LIMIT
 
 
 class TestReputationPolicy(TestCase):
@@ -17,36 +19,48 @@ class TestReputationPolicy(TestCase):
     def setUp(self):
         self.request_factory = RequestFactory()
         self.request = self.request_factory.get("/")
-        self.test_ip = "127.0.0.1"
-        self.test_username = "test"
+        self.ip = "127.0.0.1"
+        self.username = "username"
+        self.password = generate_id()
         # We need a user for the one-to-one in userreputation
-        self.user = User.objects.create(username=self.test_username)
+        self.user = User.objects.create(username=self.username)
+        self.user.set_password(self.password)
         self.backends = [BACKEND_INBUILT]
 
     def test_ip_reputation(self):
         """test IP reputation"""
         # Trigger negative reputation
-        authenticate(
-            self.request, self.backends, username=self.test_username, password=self.test_username
-        )
-        self.assertEqual(Reputation.objects.get(ip=self.test_ip).score, -1)
+        authenticate(self.request, self.backends, username=self.username, password=self.username)
+        self.assertEqual(Reputation.objects.get(ip=self.ip).score, -1)
 
     def test_user_reputation(self):
         """test User reputation"""
         # Trigger negative reputation
-        authenticate(
-            self.request, self.backends, username=self.test_username, password=self.test_username
-        )
-        self.assertEqual(Reputation.objects.get(identifier=self.test_username).score, -1)
+        authenticate(self.request, self.backends, username=self.username, password=self.username)
+        self.assertEqual(Reputation.objects.get(identifier=self.username).score, -1)
 
     def test_update_reputation(self):
         """test reputation update"""
-        Reputation.objects.create(identifier=self.test_username, ip=self.test_ip, score=43)
+        Reputation.objects.create(identifier=self.username, ip=self.ip, score=4)
         # Trigger negative reputation
-        authenticate(
-            self.request, self.backends, username=self.test_username, password=self.test_username
+        authenticate(self.request, self.backends, username=self.username, password=self.username)
+        self.assertEqual(Reputation.objects.get(identifier=self.username).score, 3)
+
+    def test_reputation_lower_limit(self):
+        """test reputation lower limit"""
+        Reputation.objects.create(identifier=self.username, ip=self.ip)
+        update_score(self.request, identifier=self.username, amount=-1000)
+        self.assertEqual(
+            Reputation.objects.get(identifier=self.username).score, DEFAULT_REPUTATION_LOWER_LIMIT
+        )
+
+    def test_reputation_upper_limit(self):
+        """test reputation upper limit"""
+        Reputation.objects.create(identifier=self.username, ip=self.ip)
+        update_score(self.request, identifier=self.username, amount=1000)
+        self.assertEqual(
+            Reputation.objects.get(identifier=self.username).score, DEFAULT_REPUTATION_UPPER_LIMIT
         )
-        self.assertEqual(Reputation.objects.get(identifier=self.test_username).score, 42)
 
     def test_policy(self):
         """Test Policy"""

authentik/tenants/api/settings.py

@@ -20,6 +20,8 @@ class Meta:
             "default_user_change_email",
             "default_user_change_username",
             "event_retention",
+            "reputation_lower_limit",
+            "reputation_upper_limit",
             "footer_links",
             "gdpr_compliance",
             "impersonation",

authentik/tenants/migrations/0005_tenant_reputation_lower_limit_and_more.py

@@ -0,0 +1,32 @@
+# Generated by Django 5.0.14 on 2025-04-14 07:50
+
+import django.core.validators
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("authentik_tenants", "0004_tenant_impersonation_require_reason"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="tenant",
+            name="reputation_lower_limit",
+            field=models.IntegerField(
+                default=-5,
+                help_text="Reputation cannot decrease lower than this value. Zero or negative.",
+                validators=[django.core.validators.MaxValueValidator(0)],
+            ),
+        ),
+        migrations.AddField(
+            model_name="tenant",
+            name="reputation_upper_limit",
+            field=models.IntegerField(
+                default=5,
+                help_text="Reputation cannot increase higher than this value. Zero or positive.",
+                validators=[django.core.validators.MinValueValidator(0)],
+            ),
+        ),
+    ]

authentik/tenants/models.py

@@ -5,7 +5,7 @@
 
 from django.apps import apps
 from django.core.exceptions import ValidationError
-from django.core.validators import MinValueValidator
+from django.core.validators import MaxValueValidator, MinValueValidator
 from django.db import models
 from django.db.utils import IntegrityError
 from django.dispatch import receiver
@@ -25,6 +25,8 @@
 
 DEFAULT_TOKEN_DURATION = "days=1"  # nosec
 DEFAULT_TOKEN_LENGTH = 60
+DEFAULT_REPUTATION_LOWER_LIMIT = -5
+DEFAULT_REPUTATION_UPPER_LIMIT = 5
 
 
 def _validate_schema_name(name):
@@ -70,6 +72,16 @@ class Tenant(TenantMixin, SerializerModel):
             "Events will be deleted after this duration.(Format: weeks=3;days=2;hours=3,seconds=2)."
         ),
     )
+    reputation_lower_limit = models.IntegerField(
+        help_text=_("Reputation cannot decrease lower than this value. Zero or negative."),
+        default=DEFAULT_REPUTATION_LOWER_LIMIT,
+        validators=[MaxValueValidator(0)],
+    )
+    reputation_upper_limit = models.IntegerField(
+        help_text=_("Reputation cannot increase higher than this value. Zero or positive."),
+        default=DEFAULT_REPUTATION_UPPER_LIMIT,
+        validators=[MinValueValidator(0)],
+    )
     footer_links = models.JSONField(
         help_text=_("The option configures the footer links on the flow executor pages."),
         default=list,

schema.yml

@@ -53723,6 +53723,17 @@ components:
           type: string
           minLength: 1
           description: 'Events will be deleted after this duration.(Format: weeks=3;days=2;hours=3,seconds=2).'
+        reputation_lower_limit:
+          type: integer
+          maximum: 0
+          minimum: -2147483648
+          description: Reputation cannot decrease lower than this value. Zero or negative.
+        reputation_upper_limit:
+          type: integer
+          maximum: 2147483647
+          minimum: 0
+          description: Reputation cannot increase higher than this value. Zero or
+            positive.
         footer_links:
           description: The option configures the footer links on the flow executor
             pages.
@@ -57775,6 +57786,17 @@ components:
         event_retention:
           type: string
           description: 'Events will be deleted after this duration.(Format: weeks=3;days=2;hours=3,seconds=2).'
+        reputation_lower_limit:
+          type: integer
+          maximum: 0
+          minimum: -2147483648
+          description: Reputation cannot decrease lower than this value. Zero or negative.
+        reputation_upper_limit:
+          type: integer
+          maximum: 2147483647
+          minimum: 0
+          description: Reputation cannot increase higher than this value. Zero or
+            positive.
         footer_links:
           description: The option configures the footer links on the flow executor
             pages.
@@ -57818,6 +57840,17 @@ components:
           type: string
           minLength: 1
           description: 'Events will be deleted after this duration.(Format: weeks=3;days=2;hours=3,seconds=2).'
+        reputation_lower_limit:
+          type: integer
+          maximum: 0
+          minimum: -2147483648
+          description: Reputation cannot decrease lower than this value. Zero or negative.
+        reputation_upper_limit:
+          type: integer
+          maximum: 2147483647
+          minimum: 0
+          description: Reputation cannot increase higher than this value. Zero or
+            positive.
         footer_links:
           description: The option configures the footer links on the flow executor
             pages.

web/src/admin/admin-settings/AdminSettingsForm.ts

@@ -23,6 +23,9 @@ import { AdminApi, FooterLink, Settings, SettingsRequest } from "@goauthentik/ap
 import "./AdminSettingsFooterLinks.js";
 import { IFooterLinkInput, akFooterLinkInput } from "./AdminSettingsFooterLinks.js";
 
+const DEFAULT_REPUTATION_LOWER_LIMIT = -5;
+const DEFAULT_REPUTATION_UPPER_LIMIT = 5;
+
 @customElement("ak-admin-settings-form")
 export class AdminSettingsForm extends Form<SettingsRequest> {
     //
@@ -177,6 +180,26 @@ export class AdminSettingsForm extends Form<SettingsRequest> {
                     <ak-utils-time-delta-help></ak-utils-time-delta-help>`}
             >
             </ak-text-input>
+            <ak-number-input
+                label=${msg("Reputation: lower limit")}
+                required
+                name="reputationLowerLimit"
+                value="${first(
+                    this._settings?.reputationLowerLimit,
+                    DEFAULT_REPUTATION_LOWER_LIMIT,
+                )}"
+                help=${msg("Reputation cannot decrease lower than this value. Zero or negative.")}
+            ></ak-number-input>
+            <ak-number-input
+                label=${msg("Reputation: upper limit")}
+                required
+                name="reputationUpperLimit"
+                value="${first(
+                    this._settings?.reputationUpperLimit,
+                    DEFAULT_REPUTATION_UPPER_LIMIT,
+                )}"
+                help=${msg("Reputation cannot increase higher than this value. Zero or positive.")}
+            ></ak-number-input>
             <ak-form-element-horizontal label=${msg("Footer links")} name="footerLinks">
                 <ak-array-input
                     .items=${this._settings?.footerLinks ?? []}