Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions null
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
fatal: no rebase in progress
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🔴 Critical

Critical: Invalid file content - Git error message instead of code.

This file contains only a Git error message (fatal: no rebase in progress) rather than actual code changes. The PR objectives indicate that guard conditions should be added to process_ip_data_and_location and update_whois_info functions to skip WHOIS lookups for deactivated devices, but no such code is present in the files provided for review.

Please verify the PR contents and ensure the actual code changes are included.

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@null` at line 1, The PR contains a Git error message instead of code —
replace the erroneous file content with the actual implementation changes and
include guard checks in the two functions mentioned: in
process_ip_data_and_location and in update_whois_info add an early-return guard
that checks the device's active/deactivated state (e.g., device.is_active,
device.status, or by loading device by device_id and checking its active flag)
and skip performing WHOIS lookups when the device is deactivated; ensure these
guards run before any network/WHOIS calls and update tests/fixtures accordingly
so the PR actually contains the modified functions rather than the git error
text.

2 changes: 1 addition & 1 deletion openwisp_controller/config/whois/commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ def clear_last_ip_command(stdout, stderr, interactive=True):
)
# include the FK field 'organization' in .only() so the related
# `organization__config_settings` traversal is not deferred
.only("last_ip", "organization", "key")
.only("last_ip", "organization", "key", "_is_deactivated")
Comment on lines 40 to +42
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🧹 Nitpick | 🔵 Trivial

Good catch preventing the N+1 from the new is_deactivated() call. Consider extending the comment.

Including _is_deactivated in .only(...) is necessary because the per-device device.save() in the loop now flows into process_ip_data_and_location()is_deactivated(), which would otherwise issue one SELECT per device for the deferred field. This is exactly what keeps test_last_ip_management_command_queries at 4 queries.

The existing comment above only motivates the organization entry; a short note about _is_deactivated would prevent someone from dropping it in a future cleanup and silently re-introducing the N+1.

♻️ Proposed comment tweak
-        # include the FK field 'organization' in .only() so the related
-        # `organization__config_settings` traversal is not deferred
+        # Include the FK 'organization' so the `organization__config_settings`
+        # traversal is not deferred, and '_is_deactivated' so the per-device
+        # save() -> process_ip_data_and_location() -> is_deactivated() path
+        # does not trigger an N+1 on the deferred field.
         .only("last_ip", "organization", "key", "_is_deactivated")
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@openwisp_controller/config/whois/commands.py` around lines 40 - 42, The
comment should be extended to explain why "_is_deactivated" is included in the
.only(...) call: because the per-device device.save() within the loop flows into
process_ip_data_and_location(), which calls is_deactivated() and would otherwise
trigger a SELECT per device for a deferred field; update the comment near the
.only("last_ip", "organization", "key", "_is_deactivated") line to note that
"_is_deactivated" is required to avoid reintroducing an N+1 query (see
is_deactivated(), process_ip_data_and_location(), and
test_last_ip_management_command_queries for context).

)
# Filter out devices that have WHOIS information for their last IP
devices = devices.exclude(last_ip=None).exclude(
Expand Down
4 changes: 4 additions & 0 deletions openwisp_controller/config/whois/service.py
Original file line number Diff line number Diff line change
Expand Up @@ -210,6 +210,8 @@ def process_ip_data_and_location(self, force_lookup=False):
Trigger WHOIS lookup based on the conditions of `_need_whois_lookup`.
Tasks are triggered on commit to ensure redundant data is not created.
"""
if self.device.is_deactivated():
return
Comment on lines +213 to +214
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🔴 Critical

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Check whether the estimated_location service also has a deactivation guard,
# since issue `#1325` requires both WHOIS and Estimated Location to be skipped.
fd -t f 'service\.py$' openwisp_controller/geo/estimated_location --exec cat {} \;
echo "---"
# Look for any is_deactivated / _is_deactivated usage inside estimated_location
rg -nP -C3 '(is_deactivated|_is_deactivated)' openwisp_controller/geo/estimated_location

Repository: openwisp/openwisp-controller

Length of output: 5427


Estimated Location service is missing the deactivation guard—fix is incomplete.

The WHOIS service guard is correct: is_deactivated() is called before _need_whois_lookup and transaction.on_commit, properly suppressing the Celery task and signal for deactivated devices.

However, issue #1325 explicitly requires the same protection for both WHOIS and Estimated Location. The EstimatedLocationService lacks an equivalent guard in trigger_estimated_location_task(). Deactivated devices will still enqueue estimated location tasks, leaving the fix incomplete.

Add an is_deactivated() guard at the entry point of EstimatedLocationService.trigger_estimated_location_task() (or the method that calls it) to match the WHOIS service implementation.

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@openwisp_controller/config/whois/service.py` around lines 213 - 214, Add the
same deactivation guard used in WHOIS: at the start of
EstimatedLocationService.trigger_estimated_location_task (or the method that
invokes it), call self.device.is_deactivated() and immediately return if True so
no Celery task or signals are enqueued; mirror the WHOIS pattern that prevents
_need_whois_lookup and transaction.on_commit from running for deactivated
devices.

new_ip = self.device.last_ip
initial_ip = self.device._initial_last_ip
if force_lookup or self._need_whois_lookup(new_ip):
Expand All @@ -229,6 +231,8 @@ def update_whois_info(self):
when the data is older than
``OPENWISP_CONTROLLER_WHOIS_REFRESH_THRESHOLD_DAYS``.
"""
if self.device.is_deactivated():
return
ip_address = self.device.last_ip
if not self.is_valid_public_ip_address(ip_address):
return
Expand Down
69 changes: 68 additions & 1 deletion openwisp_controller/config/whois/tests/tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -346,7 +346,7 @@ def test_last_ip_management_command_queries(self):
name="default.test.device4", mac_address="66:33:44:55:66:77"
)
args = ["--noinput"]
# 4 queries (3 for each device's last_ip update) and 1 for fetching devices

with self.assertNumQueries(4):
call_command("clear_last_ip", *args, stdout=out, stderr=StringIO())
Comment thread
coderabbitai[bot] marked this conversation as resolved.

Expand Down Expand Up @@ -1186,3 +1186,70 @@ def _assert_no_js_errors():
_assert_no_js_errors()
except UnexpectedAlertPresentException:
self.fail("XSS vulnerability detected in WHOIS details admin view.")


class TestWHOISDeactivated(
CreateWHOISMixin, WHOISTransactionMixin, TransactionTestCase
):
def setUp(self):
super().setUp()
self.device = self._create_device()
org_settings = self.device.organization.config_settings
org_settings.whois_enabled = True
org_settings.save()

@mock.patch.object(app_settings, "WHOIS_CONFIGURED", True)
@mock.patch("openwisp_controller.config.whois.service.fetch_whois_details.delay")
def test_process_ip_skips_when_deactivated(self, mock_task):
# Public IP that would normally trigger a lookup, so the only reason
# the task is not enqueued is the deactivation guard under test.
self.device.last_ip = "8.8.8.8"
self.device.save()
mock_task.reset_mock()

self.device.deactivate()

service = WHOISService(self.device)
service.process_ip_data_and_location()

self.assertEqual(mock_task.call_count, 0)
Comment thread
coderabbitai[bot] marked this conversation as resolved.

@mock.patch.object(app_settings, "WHOIS_CONFIGURED", True)
@mock.patch("openwisp_controller.config.whois.service.fetch_whois_details.delay")
def test_process_ip_runs_when_active(self, mock_task):
self.device.last_ip = "8.8.8.8"
self.device.save()

mock_task.reset_mock()

service = WHOISService(self.device)
service.process_ip_data_and_location()

# transaction.on_commit executes immediately in TransactionTestCase,
# so the task is triggered synchronously here
mock_task.assert_called_once_with(
device_pk=self.device.pk,
initial_ip_address=self.device._initial_last_ip,
)
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Comment thread
coderabbitai[bot] marked this conversation as resolved.

@mock.patch.object(app_settings, "WHOIS_CONFIGURED", True)
@mock.patch("openwisp_controller.config.whois.service.fetch_whois_details.delay")
def test_update_whois_skips_when_deactivated(self, mock_task):
ip = "8.8.8.8"
self.device.last_ip = ip
self.device.save()

threshold = app_settings.WHOIS_REFRESH_THRESHOLD_DAYS + 1
expired_time = timezone.now() - timedelta(days=threshold)

whois_obj = self._create_whois_info(ip_address=ip)
WHOISInfo.objects.filter(pk=whois_obj.pk).update(modified=expired_time)

self.device.deactivate()

mock_task.reset_mock()

service = WHOISService(self.device)
service.update_whois_info()
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

CRITICAL: Variable name typo - ervice should be service. This causes a NameError at runtime since service is undefined.

Suggested change
service.update_whois_info()
service = WHOISService(self.device)


mock_task.assert_not_called()
Comment on lines +1235 to +1255
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

test_update_whois_skips_when_deactivated is not a valid regression guard.

Even after fixing the ervice typo, this test would pass with or without the deactivation guard in update_whois_info:

  1. No WHOISInfo row exists for 8.8.8.8, so update_whois_info() hits whois_obj = ...first()None and returns without ever scheduling fetch_whois_details.delay — regardless of deactivation.
  2. self.device.deactivate() typically clears last_ip/management_ip, so by the time update_whois_info() runs, self.is_valid_public_ip_address(self.device.last_ip) is also false — another skip path unrelated to the guard.

To actually pin the if self.device.is_deactivated(): return added in service.py (lines 234–235), create an expired WHOISInfo for the IP and assert that delay is called on an active device in a sibling test, then assert it is not called after deactivate() — mirroring the active/deactivated pair you already have for process_ip_data_and_location.

🧪 Sketch of a stronger regression pair
     `@mock.patch.object`(app_settings, "WHOIS_CONFIGURED", True)
     `@mock.patch`("openwisp_controller.config.whois.service.fetch_whois_details.delay")
     def test_update_whois_skips_when_deactivated(self, mock_task):
-        self.device.last_ip = "8.8.8.8"
-        self.device.save()
-        mock_task.reset_mock()
-
-        self.device.deactivate()
-
-        ervice = WHOISService(self.device)
-        service.update_whois_info()
-
-        mock_task.assert_not_called()
+        ip = "8.8.8.8"
+        self.device.last_ip = ip
+        self.device.save()
+        # Make an expired WHOISInfo so update_whois_info() would normally
+        # schedule a refresh; the deactivation guard is the only reason it won't.
+        threshold = app_settings.WHOIS_REFRESH_THRESHOLD_DAYS + 1
+        expired = timezone.now() - timedelta(days=threshold)
+        whois_obj = self._create_whois_info(ip_address=ip)
+        WHOISInfo.objects.filter(pk=whois_obj.pk).update(modified=expired)
+        self.device.deactivate()
+        mock_task.reset_mock()
+
+        service = WHOISService(self.device)
+        service.update_whois_info()
+
+        mock_task.assert_not_called()
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@openwisp_controller/config/whois/tests/tests.py` around lines 1233 - 1245,
The test test_update_whois_skips_when_deactivated is invalid because it never
exercises the deactivation branch; fix it by creating a WHOISInfo row for the
device IP that is expired (so update_whois_info would attempt to schedule work),
correct the typo "ervice" → "service", then assert fetch_whois_details.delay is
called for the active device and after calling self.device.deactivate() call
WHOISService(self.device).update_whois_info() again and assert delay is not
called; target WHOISService.update_whois_info, the WHOISInfo test
fixture/creation, and the mocked fetch_whois_details.delay in your changes.

2 changes: 2 additions & 0 deletions openwisp_controller/geo/estimated_location/service.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,8 @@ def is_estimated_location_enabled(self):
return self.check_estimated_location_enabled(self.device.organization_id)

def trigger_estimated_location_task(self, ip_address):
if self.device.is_deactivated():
return
try:
current_app.send_task(
"whois_estimated_location_task",
Expand Down
Loading