Fixes skupperproject#1225: prevent management deletion of tcp adaptor…

… egress dispatch connection (skupperproject#1227)

* Fixes skupperproject#1225: prevent management from deleting egress-dispatcher connections

* fixup: add tcp and edge test cases

* fixup: move test earlier in the test sequence

src/router_core/agent_connection.c

@@ -524,9 +524,12 @@ static void qdra_connection_update_set_status(qdr_core_t *core, qdr_query_t *que
         qd_iterator_t *admin_status_iter = qd_parse_raw(admin_state);
 
         if (qd_iterator_equal(admin_status_iter, (unsigned char*) QDR_CONNECTION_ADMIN_STATUS_DELETED)) {
-            // This connection has been force-closed.
-            // Inter-router and edge connections may not be force-closed
-            if (conn->role != QDR_ROLE_INTER_ROUTER && conn->role != QDR_ROLE_EDGE_CONNECTION) {
+            // This connection has been force-closed.  There are certain types of connections that cannot be
+            // force-closed by management because they are essential for the correct operation of the router
+            if (conn->role != QDR_ROLE_INTER_ROUTER
+                && conn->role != QDR_ROLE_EDGE_CONNECTION
+                // ISSUE-1225: cannot delete tcp dispatch connections
+                && (!conn->connection_info->host || strcmp(conn->connection_info->host, "egress-dispatch"))) {
                 qdr_close_connection_CT(core, conn);
                 qd_log(LOG_AGENT,
                        QD_LOG_INFO,

tests/system_tests_edge_router.py

@@ -29,11 +29,13 @@
 from proton.utils import BlockingConnection
 
 from skupper_router.management.client import Node
+from skupper_router.management.error import ForbiddenStatus
 
 from system_test import TestCase, Qdrouterd, main_module, TIMEOUT, MgmtMsgProxy, TestTimeout
 from system_test import QdManager
 from system_test import unittest
 from system_test import Process
+from system_test import CONNECTION_TYPE
 from test_broker import FakeBroker
 
 from message_tests import DynamicAddressTest, MobileAddressAnonymousTest, MobileAddressTest
@@ -351,7 +353,7 @@ def router(name, mode, connection, extra=None):
                     'test_73':  0
                     }
 
-    def test_01_connectivity_INTA_EA1(self):
+    def test_01_1_connectivity_INTA_EA1(self):
         if self.skip['test_01'] :
             self.skipTest("Test skipped during development.")
 
@@ -361,6 +363,31 @@ def test_01_connectivity_INTA_EA1(self):
         test.run()
         self.assertIsNone(test.error)
 
+    def test_01_2_ensure_no_admin_delete(self):
+        """
+        This test tries to delete edge uplink connections but that operation
+        is forbidden. This test runs immediately after the INTA/EA1
+        connectivity test to ensure both routers are operational. Futher tests
+        of the INTA/EA1 path follow to ensure the delete attempt did not change
+        the routers state.
+        """
+        if self.skip['test_01'] :
+            self.skipTest("Test skipped during development.")
+
+        mgmt_a = self.routers[0].management
+        mgmt_ea1 = self.routers[2].management
+
+        for mgmt in (mgmt_a, mgmt_ea1):
+            conns = [d for d in
+                     mgmt.query(type=CONNECTION_TYPE).get_dicts() if
+                     d['role'] == 'edge']
+            self.assertNotEqual(0, len(conns), f"Expected at least one connection: {conns}")
+            for conn in conns:
+                with self.assertRaises(ForbiddenStatus):
+                    mgmt.update(attributes={'adminStatus': 'deleted'},
+                                type=CONNECTION_TYPE,
+                                identity=conn['identity'])
+
     def test_02_connectivity_INTA_EA2(self):
         if self.skip['test_02'] :
             self.skipTest("Test skipped during development.")

tests/system_tests_tcp_adaptor.py

@@ -32,6 +32,8 @@
 from proton.handlers import MessagingHandler
 from proton.reactor import Container
 
+from skupper_router.management.error import ForbiddenStatus
+
 from system_test import Logger
 from system_test import Process
 from system_test import Qdrouterd
@@ -1733,6 +1735,17 @@ def _wait_for_close():
             self.assertEqual(2, c_stats['connectionsOpened'])
             self.assertEqual(1, c_stats['connectionsClosed'])
 
+            # Attempt to delete the special tcp-dispatcher pseudo connection.
+            # Doing so is forbidden so expect this to fail
+
+            conns = [d for d in mgmt.query(type=CONNECTION_TYPE).get_dicts()
+                     if d['host'] == 'egress-dispatch']
+            for conn in conns:
+                with self.assertRaises(ForbiddenStatus):
+                    mgmt.update(attributes={'adminStatus': 'deleted'},
+                                type=CONNECTION_TYPE,
+                                identity=conn['identity'])
+
             # splendid!  Not delete all the things
 
             mgmt.delete(type=LISTENER_TYPE, name=listener_name)