Added remote port (jazzband#671)

Author: CleitonDeLima

.gitignore

@@ -80,3 +80,4 @@ venv.bak/
 
 ### JetBrains
 .idea/
+.vscode/

CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ## Next Release
 
+#### Improvements
+- feat: Added `LogEntry.remote_port` field. ([#671](https://github.com/jazzband/django-auditlog/pull/671))
+
 #### Fixes
 
 - Fixed problem when setting `django.db.models.functions.Now()` in `DateTimeField` ([#635](https://github.com/jazzband/django-auditlog/pull/635))

auditlog/context.py

@@ -13,12 +13,13 @@
 
 
 @contextlib.contextmanager
-def set_actor(actor, remote_addr=None):
+def set_actor(actor, remote_addr=None, remote_port=None):
     """Connect a signal receiver with current user attached."""
     # Initialize thread local storage
     context_data = {
         "signal_duid": ("set_actor", time.time()),
         "remote_addr": remote_addr,
+        "remote_port": remote_port,
     }
     auditlog_value.set(context_data)
 
@@ -63,6 +64,7 @@ def _set_actor(user, sender, instance, signal_duid, **kwargs):
             instance.actor = user
 
         instance.remote_addr = auditlog["remote_addr"]
+        instance.remote_port = auditlog["remote_port"]
 
 
 @contextlib.contextmanager

auditlog/middleware.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from django.conf import settings
 from django.contrib.auth import get_user_model
 
@@ -36,6 +38,17 @@ def _get_remote_addr(request):
 
         return remote_addr
 
+    @staticmethod
+    def _get_remote_port(request) -> Optional[int]:
+        remote_port = request.headers.get("X-Forwarded-Port", "")
+
+        try:
+            remote_port = int(remote_port)
+        except ValueError:
+            remote_port = None
+
+        return remote_port
+
     @staticmethod
     def _get_actor(request):
         user = getattr(request, "user", None)
@@ -45,9 +58,10 @@ def _get_actor(request):
 
     def __call__(self, request):
         remote_addr = self._get_remote_addr(request)
+        remote_port = self._get_remote_port(request)
         user = self._get_actor(request)
 
         set_cid(request)
 
-        with set_actor(actor=user, remote_addr=remote_addr):
+        with set_actor(actor=user, remote_addr=remote_addr, remote_port=remote_port):
             return self.get_response(request)

auditlog/migrations/0018_logentry_remote_port.py

@@ -0,0 +1,17 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("auditlog", "0017_alter_logentry_changes"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="logentry",
+            name="remote_port",
+            field=models.PositiveIntegerField(
+                blank=True, null=True, verbose_name="remote port"
+            ),
+        ),
+    ]

auditlog/models.py

@@ -358,6 +358,9 @@ class Action:
     remote_addr = models.GenericIPAddressField(
         blank=True, null=True, verbose_name=_("remote address")
     )
+    remote_port = models.PositiveIntegerField(
+        blank=True, null=True, verbose_name=_("remote port")
+    )
     timestamp = models.DateTimeField(
         default=django_timezone.now,
         db_index=True,

auditlog_tests/tests.py

@@ -537,6 +537,13 @@ def test_get_remote_addr(self):
                     self.middleware._get_remote_addr(request), expected_remote_addr
                 )
 
+    def test_get_remote_port(self):
+        headers = {
+            "HTTP_X_FORWARDED_PORT": "12345",
+        }
+        request = self.factory.get("/", **headers)
+        self.assertEqual(self.middleware._get_remote_port(request), 12345)
+
     def test_cid(self):
         header = str(settings.AUDITLOG_CID_HEADER).lstrip("HTTP_").replace("_", "-")
         header_meta = "HTTP_" + header.upper().replace("-", "_")
@@ -574,9 +581,10 @@ def test_set_actor_anonymous_request(self):
         The remote address will be set even when there is no actor
         """
         remote_addr = "123.213.145.99"
+        remote_port = 12345
         actor = None
 
-        with set_actor(actor=actor, remote_addr=remote_addr):
+        with set_actor(actor=actor, remote_addr=remote_addr, remote_port=remote_port):
             obj = SimpleModel.objects.create(text="I am not difficult.")
 
             history = obj.history.get()
@@ -585,6 +593,11 @@ def test_set_actor_anonymous_request(self):
                 remote_addr,
                 msg=f"Remote address is {remote_addr}",
             )
+            self.assertEqual(
+                history.remote_port,
+                remote_port,
+                msg=f"Remote port is {remote_port}",
+            )
             self.assertIsNone(history.actor, msg="Actor is `None` for anonymous user")
 
     def test_get_actor(self):