From 1f14c42af554f1a14124257843b9fc1a0a06755f Mon Sep 17 00:00:00 2001
From: Sunil Mohan Adapa <sunil@medhas.org>
Date: Wed, 18 Dec 2019 12:09:12 -0800
Subject: [PATCH] firewall: Implement new diagnostic tests to check port status

- Test that a port meant to available to internal network is available.

- Test that a port meant to available to external network is available.

- Test that a port meant to unavailable to external network is unavailable.

Signed-off-by: Sunil Mohan Adapa <sunil@medhas.org>
Reviewed-by: James Valleroy <jvalleroy@mailbox.org>
---
 plinth/modules/firewall/components.py         | 37 ++++++++++++
 .../modules/firewall/tests/test_components.py | 56 +++++++++++++++++++
 2 files changed, 93 insertions(+)

diff --git a/plinth/modules/firewall/components.py b/plinth/modules/firewall/components.py
index 8d0a7040c..5af546f73 100644
--- a/plinth/modules/firewall/components.py
+++ b/plinth/modules/firewall/components.py
@@ -20,6 +20,8 @@ App component for other apps to use firewall functionality.
 
 import logging
 
+from django.utils.translation import ugettext_lazy as _
+
 from plinth import app
 from plinth.modules import firewall
 
@@ -109,3 +111,38 @@ class Firewall(app.FollowerComponent):
     def get_internal_interfaces():
         """Returns a list of interfaces in a firewall zone."""
         return firewall.get_interfaces('internal')
+
+    def diagnose(self):
+        """Check if the firewall ports are open and only as expected.
+
+        See :py:meth:`plinth.app.Component.diagnose`.
+
+        """
+        results = []
+        internal_ports = firewall.get_enabled_services(zone='internal')
+        external_ports = firewall.get_enabled_services(zone='external')
+        for port_detail in self.ports_details:
+            port = port_detail['name']
+
+            # Internal zone
+            result = 'passed' if port in internal_ports else 'failed'
+            message = _(
+                'Port {name} ({details}) available for internal networks'
+            ).format(name=port, details=port_detail['details'])
+            results.append([message, result])
+
+            # External zone
+            if self.is_external:
+                result = 'passed' if port in external_ports else 'failed'
+                message = _(
+                    'Port {name} ({details}) available for external networks'
+                ).format(name=port, details=port_detail['details'])
+            else:
+                result = 'passed' if port not in external_ports else 'failed'
+                message = _(
+                    'Port {name} ({details}) unavailable for external networks'
+                ).format(name=port, details=port_detail['details'])
+
+            results.append([message, result])
+
+        return results
diff --git a/plinth/modules/firewall/tests/test_components.py b/plinth/modules/firewall/tests/test_components.py
index cf5bac265..dc4e90969 100644
--- a/plinth/modules/firewall/tests/test_components.py
+++ b/plinth/modules/firewall/tests/test_components.py
@@ -136,3 +136,59 @@ def test_disable(get_enabled_services, add_service, remove_service):
         call('test-port4', zone='external')
     ]
     remove_service.assert_has_calls(calls)
+
+
+@patch('plinth.modules.firewall.get_port_details')
+@patch('plinth.modules.firewall.get_enabled_services')
+def test_diagnose(get_enabled_services, get_port_details):
+    """Test diagnosing open/closed firewall ports."""
+    def get_port_details_side_effect(port):
+        return {
+            'test-port1': '1234/tcp',
+            'test-port2': '2345/udp',
+            'test-port3': '3456/tcp',
+            'test-port4': '4567/udp'
+        }[port]
+
+    def get_enabled_services_side_effect(zone):
+        return {
+            'internal': ['test-port1', 'test-port3'],
+            'external': ['test-port2', 'test-port3']
+        }[zone]
+
+    get_enabled_services.side_effect = get_enabled_services_side_effect
+    get_port_details.side_effect = get_port_details_side_effect
+    firewall = Firewall('test-firewall-1', ports=['test-port1', 'test-port2'],
+                        is_external=False)
+    results = firewall.diagnose()
+    assert results == [
+        [
+            'Port test-port1 (1234/tcp) available for internal networks',
+            'passed'
+        ],
+        [
+            'Port test-port1 (1234/tcp) unavailable for external networks',
+            'passed'
+        ],
+        [
+            'Port test-port2 (2345/udp) available for internal networks',
+            'failed'
+        ],
+        [
+            'Port test-port2 (2345/udp) unavailable for external networks',
+            'failed'
+        ]
+    ]
+
+    firewall = Firewall('test-firewall-1', ports=['test-port3', 'test-port4'],
+                        is_external=True)
+    results = firewall.diagnose()
+    assert results == [[
+        'Port test-port3 (3456/tcp) available for internal networks', 'passed'
+    ], [
+        'Port test-port3 (3456/tcp) available for external networks', 'passed'
+    ], [
+        'Port test-port4 (4567/udp) available for internal networks', 'failed'
+    ], [
+        'Port test-port4 (4567/udp) available for external networks', 'failed'
+    ]]