core: services: cable_guy: Move interface priority from ipr to dhcpcd

core/frontend/src/components/app/NetworkInterfacePriorityMenu.vue

@@ -3,7 +3,7 @@
     <v-list>
       <v-card-subtitle class="text-md-center" max-width="30">
         Move network interfaces over
-        to change network access priority
+        to change network access priority.<br>Applied changes require a <b>system reboot</b>.
       </v-card-subtitle>
       <draggable v-model="interfaces">
         <v-card
@@ -30,6 +30,13 @@
     </v-list>
     <v-divider />
     <v-card-actions class="justify-center pa-2">
+      <v-btn
+        color="primary"
+        @click="close()"
+      >
+        Cancel
+      </v-btn>
+      <v-spacer />
       <v-btn
         color="success"
         :disabled="is_loading"
@@ -76,6 +83,7 @@ export default Vue.extend({
       return `${index}${['st', 'nd', 'rd'][((index + 90) % 100 - 10) % 10 - 1] || 'th'}`
     },
     async setHighestInterface(): Promise<void> {
+      this.is_loading = true
       const interface_priority = this.interfaces.map((inter) => ({ name: inter.name }))
       await back_axios({
         method: 'post',
@@ -95,6 +103,8 @@ export default Vue.extend({
           )
           this.close()
         })
+      await this.fetchAvailableInterfaces()
+      this.is_loading = false
     },
     async fetchAvailableInterfaces(): Promise<void> {
       await back_axios({
@@ -105,9 +115,9 @@ export default Vue.extend({
         .then((response) => {
           const interfaces = response.data as EthernetInterface[]
           interfaces.sort((a, b) => {
-            if (!a.info) return 1
-            if (!b.info) return -1
-            return b.info.priority - a.info.priority
+            if (!a.info) return -1
+            if (!b.info) return 1
+            return a.info.priority - b.info.priority
           })
           this.interfaces = interfaces
         })

core/services/cable_guy/api/manager.py

@@ -60,9 +60,16 @@ class EthernetManager:
     # DNS abstraction
     dns = dns.Dns()
 
+    # https://man.archlinux.org/man/dhcpcd.conf.5#metric
+    default_dhcpdc_metric = 1000
+
     result: List[NetworkInterface] = []
 
     def __init__(self, default_configs: List[NetworkInterface]) -> None:
+        self.dhcpcd_conf_path = "/etc/dhcpcd.conf"
+        self.dhcpcd_conf_start_string = "#blueos-interface-priority-start"
+        self.dhcpcd_conf_end_string = "#blueos-interface-priority-end"
+
         self.settings = settings.Settings()
 
         self._dhcp_servers: List[DHCPServerManager] = []
@@ -365,8 +372,20 @@ def get_interface_ndb(self, interface_name: str) -> Any:
         """
         return self.ndb.interfaces.dump().filter(ifname=interface_name)[0]
 
-    @temporary_cache(timeout_seconds=5)
+    @temporary_cache(timeout_seconds=1)
     def get_interfaces_priority(self) -> List[NetworkInterfaceMetric]:
+        """Get priority of network interfaces dhcpcd otherwise fetch from ipr.
+
+        Returns:
+            List[NetworkInterfaceMetric]: List of interface priorities, lower is higher priority
+        """
+        result = self._get_interface_priority_from_dhcpcd()
+        if result:
+            return result
+
+        return self._get_interfaces_priority_from_ipr()
+
+    def _get_interfaces_priority_from_ipr(self) -> List[NetworkInterfaceMetric]:
         """Get the priority metrics for all network interfaces.
 
         Returns:
@@ -396,71 +415,182 @@ def get_interfaces_priority(self) -> List[NetworkInterfaceMetric]:
             else:
                 metric_dict[d["index"]] = d["metric"]
 
+        # Highest priority wins for ipr but not for dhcpcd, so we sort and reverse the list
+        # Where we change the priorities between highest and low to convert that
+        original_list = sorted(
+            [
+                NetworkInterfaceMetric(index=index, name=name, priority=metric_dict.get(index) or 0)
+                for index, name in name_dict.items()
+            ],
+            key=lambda x: x.priority,
+            reverse=True,
+        )
+
         return [
-            NetworkInterfaceMetric(index=index, name=name, priority=metric_dict.get(index) or 0)
-            for index, name in name_dict.items()
+            NetworkInterfaceMetric(index=item.index, name=item.name, priority=original_list[-(i + 1)].priority)
+            for i, item in enumerate(original_list)
         ]
 
-    def get_interface_priority(self, interface_name: str) -> Optional[NetworkInterfaceMetric]:
-        """Get the priority metric for a network interface.
+    def _get_service_dhcpcd_content(self) -> List[str]:
+        """Returns a list of lines from the dhcpcd configuration file that belong to
+        this service.
+        Any exceptions are caught and logged, and an empty list is returned.
 
-        Args:
-            interface_name (str): The name of the network interface.
+        List[str]: Lines that will be used by this service
+        """
+        try:
+            with open(self.dhcpcd_conf_path, "r", encoding="utf-8") as f:
+                lines = f.readlines()
+
+                start, end = None, None
+                for i, line in enumerate(lines):
+                    # Get always the first occurrence of 'start' and last of 'end'
+                    if self.dhcpcd_conf_start_string in line and start is None:
+                        start = i
+                    if self.dhcpcd_conf_end_string in line:
+                        end = i
+
+                # Remove everything that is not from us
+                if start is not None and end is not None:
+                    del lines[0 : start + 1]
+                    del lines[end:-1]
+
+                # Clean all lines and remove empty ones
+                lines = [line.strip() for line in lines]
+                lines = [line for line in lines if line]
+                return lines
+        except Exception as exception:
+            logger.warning(f"Failed to read {self.dhcpcd_conf_path}, error: {exception}")
+            return []
 
-        Returns:
-            Optional[NetworkInterfaceMetric]: The priority metric for the interface, or None if no metric found.
+    def _get_interface_priority_from_dhcpcd(self) -> List[NetworkInterfaceMetric]:
+        """Parses dhcpcd config file to get network interface priorities.
+        Goes through the dhcpcd config file line by line looking for "interface"
+        and "metric" lines. Extracts the interface name and metric value. The
+        metric is used as the priority, with lower being better.
+
+        List[NetworkInterfaceMetric]: A list of priority metrics for each interface.
         """
-        metric: NetworkInterfaceMetric
-        for metric in self.get_interfaces_priority():
-            if interface_name == metric.name:
-                return metric
+        lines = self._get_service_dhcpcd_content()
+        result = []
+        current_interface = None
+        current_metric = None
+        for line in lines:
+            if line.startswith("interface"):
+                if current_interface is not None and current_metric is not None:  # type: ignore[unreachable]
+                    # Metric is inverted compared to priority, lowest metric wins
+                    result.append(NetworkInterfaceMetric(index=0, name=current_interface, priority=current_metric))  # type: ignore[unreachable]
+
+                current_interface = line.split()[1]
+                current_metric = None
+
+            elif line.startswith("metric") and current_interface is not None:
+                try:
+                    current_metric = int(line.split()[1])
+                except Exception as exception:
+                    logger.error(
+                        f"Failed to parse {current_interface} metric, error: {exception}, line: {line}, using default metric"
+                    )
+                    current_metric = EthernetManager.default_dhcpdc_metric
+
+        # Add the last entry to the result_list
+        if current_interface is not None and current_metric is not None:
+            result.append(NetworkInterfaceMetric(index=0, name=current_interface, priority=current_metric))
 
-        return None
+        return result
+
+    def _remove_dhcpcd_configuration(self) -> None:
+        """Removes the network priority configuration added by this service from
+        dhcpcd.conf file.
+        """
+        lines = []
+        with open(self.dhcpcd_conf_path, "r", encoding="utf-8") as f:
+            lines = f.readlines()
+
+            start, end = None, None
+            for i, line in enumerate(lines):
+                # Get always the first occurrence of 'start' and last of 'end'
+                if self.dhcpcd_conf_start_string in line and start is None:
+                    start = i
+                if self.dhcpcd_conf_end_string in line:
+                    end = i
+
+            # Remove our part
+            if start is not None and end is not None:
+                logger.info(f"Deleting rage: {start} : {end + 1}")
+                del lines[start : end + 1]
+            else:
+                logger.info(f"There is no network priority configuration in {self.dhcpcd_conf_path}")
+                return
+
+        if not lines:
+            logger.warning(f"{self.dhcpcd_conf_path} appears to be empty.")
+            return
+
+        with open("/etc/dhcpcd.conf", "w", encoding="utf-8") as f:
+            f.writelines(lines)
 
     def set_interfaces_priority(self, interfaces: List[NetworkInterfaceMetricApi]) -> None:
-        """Set interfaces priority.
+        """Sets network interface priority. This is an abstraction function for different
+        implementations.
+
+        Args:
+            interfaces (List[NetworkInterfaceMetricApi]): A list of interfaces and their priority metrics.
+                sorted by priority to set, if values are undefined.
+        """
+        self._set_interfaces_priority_to_dhcpcd(interfaces)
+
+    def _set_interfaces_priority_to_dhcpcd(self, interfaces: List[NetworkInterfaceMetricApi]) -> None:
+        """Sets network interface priority..
 
         Args:
             interfaces (List[NetworkInterfaceMetricApi]): A list of interfaces and their priority metrics.
         """
+
+        # Note: With DHCPCD, lower priority wins!
+        self._remove_dhcpcd_configuration()
+
+        # Update interfaces priority if possible
         if not interfaces:
             logger.info("Cant change network priority from empty list.")
             return
 
-        highest_metric = 1
-        for interface in self.get_interfaces_priority():
-            highest_metric = max(highest_metric, interface.priority)
-
         # If there is a single interface without metric, make it the highest priority
         if len(interfaces) == 1 and interfaces[0].priority is None:
-            interface = self.get_interface_priority(interfaces[0].name)
-            original_priority = interface and interface.priority or 0
-            for interface in self.get_interfaces_priority():
-                highest_metric = max(highest_metric, original_priority)
-            if original_priority == highest_metric:
-                logger.info(f"Interface {interfaces[0].name} already has highest priority: {highest_metric}")
-                return
-            interfaces[0].priority = highest_metric + 10
+            interfaces[0].priority = 0
 
-        # Ensure a high value for metric
-        if highest_metric <= len(interfaces):
-            highest_metric = 400
+        current_priority = interfaces[0].priority or EthernetManager.default_dhcpdc_metric
+        lines = []
+        lines.append(f"{self.dhcpcd_conf_start_string}\n")
+        for interface in interfaces:
+            # Enforce priority if it's none, otherwise track new priority
+            interface.priority = interface.priority or current_priority
+            current_priority = interface.priority
 
-        if all(interface.priority is not None for interface in interfaces):
-            for interface in interfaces:
-                EthernetManager.set_interface_priority(interface.name, interface.priority)
-            return
+            lines.append(f"interface {interface.name}\n")
+            lines.append(f"    metric {interface.priority}\n")
+            current_priority += 1000
+            logger.info(f"Set current priority for {interface.name} as {interface.priority}")
+        lines.append(f"{self.dhcpcd_conf_end_string}\n")
 
-        # Calculate metric automatically in the case where no metric is provided
-        if all(interface.priority is None for interface in interfaces):
-            network_step = int(highest_metric / len(interfaces))
-            times = 0
-            for interface in interfaces:
-                EthernetManager.set_interface_priority(interface.name, highest_metric - network_step * times)
-                times += 1
-            return
+        with open("/etc/dhcpcd.conf", "a+", encoding="utf-8") as f:
+            f.writelines(lines)
 
-        raise RuntimeError("There is no support for interfaces with and without metric on the same list")
+    def get_interface_priority(self, interface_name: str) -> Optional[NetworkInterfaceMetric]:
+        """Get the priority metric for a network interface.
+
+        Args:
+            interface_name (str): The name of the network interface.
+
+        Returns:
+            Optional[NetworkInterfaceMetric]: The priority metric for the interface, or None if no metric found.
+        """
+        metric: NetworkInterfaceMetric
+        for metric in self.get_interfaces_priority():
+            if interface_name == metric.name:
+                return metric
+
+        return None
 
     @staticmethod
     def set_interface_priority(name: str, priority: int) -> None:

core/services/cable_guy/main.py

@@ -80,7 +80,7 @@ def configure_interface(interface: NetworkInterface = Body(...)) -> Any:
 
 @app.get("/interfaces", response_model=List[NetworkInterface], summary="Retrieve all network interfaces.")
 @version(1, 0)
-@temporary_cache(timeout_seconds=10)
+@temporary_cache(timeout_seconds=1)
 def retrieve_interfaces() -> Any:
     """REST API endpoint to retrieve the all network interfaces."""
     return manager.get_interfaces()