vpnaas: Add swanctl (VICI protocol) support for strongSwan

Add support for ``swanctl`` as an alternative to the legacy
stroke/starter interface in the strongSwan device driver. This enables
compatibility with modern Linux distributions (Fedora, RHEL 9+,
recent Ubuntu/Debian) that ship strongSwan without the deprecated
stroke plugin.

Key changes:

- Added ``use_swanctl`` configuration option under ``[strongswan]``
  to enable swanctl mode (default: False for backward compatibility).

- Implemented new ``SwanctlProcess`` class that manages strongSwan via
  the swanctl CLI frontend using the VICI protocol. This replaces the
  legacy ``ipsec start/stop/stroke/reload`` commands with their swanctl
  equivalents:

  * ``swanctl --initiate --child <id>`` instead of ``ipsec stroke
    up-nb <id>``
  * ``swanctl --load-all`` instead of ``ipsec reload``
  * ``swanctl --load-creds`` instead of ``ipsec rereadsecrets``
  * ``swanctl --list-sas`` instead of ``ipsec status``

- Added systemd-based daemon management for charon (strongSwan IKE
  daemon), replacing the legacy ipsec starter approach.

- Created new Jinja2 templates for swanctl configuration:
  * ``template/swanctl/swanctl.conf.template`` - TOML-like connection
    config
  * ``template/swanctl/ipsec.secrets.template`` - secrets in swanctl
    format

- Implemented status parsing for ``swanctl --list-sas`` output format,
  which uses indented CHILD_SA blocks with states like INSTALLED,
  CONNECTING, and REKEYING.

- Added ``OvnSwanctlDriver`` class for OVN-based VPN deployments using
  swanctl.

- Deprecated ``StrongSwanProcess`` with a warning message directing
  users to migrate to ``SwanctlProcess`` via the ``use_swanctl``
  option.

Closes-Bug: #2160165
Assisted-By: Composer 2.5
Signed-off-by: Rodolfo Alonso Hernandez <ralonsoh@redhat.com>
Change-Id: I8f3b7c9d2e1a4b5c6d7e8f9a0b1c2d3e4f5a6b7c

neutron_vpnaas/services/vpn/device_drivers/ovn_ipsec.py

@@ -346,6 +346,19 @@ class OvnStrongSwanDriver(OvnIPsecDriver):
             namespace)
 
 
+class OvnSwanctlDriver(OvnIPsecDriver):
+    """OVN IPsec driver using swanctl (VICI protocol).
+
+    .. versionadded:: 25.1.0
+    """
+    def create_process(self, process_id, vpnservice, namespace):
+        return strongswan_ipsec.SwanctlProcess(
+            self.conf,
+            process_id,
+            vpnservice,
+            namespace)
+
+
 class OvnLibreSwanDriver(OvnIPsecDriver):
     def create_process(self, process_id, vpnservice, namespace):
         return libreswan_ipsec.LibreSwanProcess(

neutron_vpnaas/services/vpn/device_drivers/strongswan_ipsec.py

@@ -14,13 +14,17 @@
 #    under the License.
 
 import os
-
-from oslo_config import cfg
-from oslo_log import log as logging
+import re
+import time
+import warnings
 
 from neutron.agent.linux import ip_lib
 from neutron.agent.linux import utils
 from neutron_lib import constants
+from neutron_lib.utils import file as file_utils
+from oslo_config import cfg
+from oslo_log import log as logging
+from oslo_utils import fileutils
 
 from neutron_vpnaas._i18n import _
 from neutron_vpnaas.services.vpn.device_drivers import ipsec
@@ -51,7 +55,36 @@ strongswan_opts = [
         'default_config_area',
         default='/etc/strongswan.d',
         help=_('The area where default StrongSwan configuration '
-               'files are located.'))
+               'files are located.')),
+    # swanctl options
+    cfg.BoolOpt(
+        'use_swanctl',
+        default=True,
+        deprecated_since='2027.1',
+        deprecated_reason=_('The use_swanctl option is deprecated and will be '
+                            'removed in the 2027.2 release. ``swanctl`` mode '
+                            'is the default behavior.'),
+        help=_('Use ``swanctl`` (VICI protocol) instead of legacy stroke '
+               'interface. When True, configuration files are generated in '
+               '/etc/swanctl/ and charon is managed via systemd.')),
+    cfg.StrOpt(
+        'swanctl_config_template',
+        default=os.path.join(
+            TEMPLATE_PATH,
+            'template/swanctl/swanctl.conf.template'),
+        help=_('Template file for swanctl configuration.')),
+    cfg.StrOpt(
+        'swanctl_secrets_template',
+        default=os.path.join(
+            TEMPLATE_PATH,
+            'template/swanctl/ipsec.secrets.template'),
+        help=_('Template file for swanctl secrets configuration.')),
+    cfg.StrOpt(
+        'swanctl_config_dir',
+        default='/etc/swanctl',
+        help=_('Directory where swanctl loads its configuration from. '
+               'This directory must contain swanctl.conf, connections/, '
+               'secrets/.'))
 ]
 cfg.CONF.register_opts(strongswan_opts, 'strongswan')
 
@@ -74,6 +107,11 @@ class StrongSwanProcess(ipsec.BaseSwanProcess):
     STATUS_NOT_RUNNING_RE = 'Command:.*ipsec.*status.*Exit code: [1|3] '
 
     def __init__(self, conf, process_id, vpnservice, namespace):
+        warnings.warn(
+            "StrongSwanProcess is deprecated and will be removed in a future "
+            "release. Use SwanctlProcess with use_swanctl=True instead.",
+            DeprecationWarning,
+            stacklevel=2)
         dialect_map_update = {
             'v1': 'ikev1',
             'v2': 'ikev2',
@@ -242,6 +280,319 @@ class StrongSwanProcess(ipsec.BaseSwanProcess):
         self.connection_status = {}
 
 
+class SwanctlProcess(ipsec.BaseSwanProcess):
+    """swanctl-based strongSwan process using VICI protocol.
+
+    This class manages strongSwan via the swanctl CLI frontend which uses
+    the modern VICI protocol instead of the deprecated stroke/starter
+    interface.
+
+    .. versionadded:: 25.1.0
+    """
+
+    # States from swanctl --list-sas output
+    STATUS_DICT = {
+        'INSTALLED': constants.ACTIVE,
+        'CONNECTING': constants.DOWN,
+        'REKEYING': constants.ACTIVE,
+        'DELETING': constants.DOWN,
+    }
+    # Match lines like: "child_sa:  VPN_abc123_child_1" followed by state on
+    # next line
+    STATUS_RE = r'(INSTALLED|CONNECTING|REKEYING|DELETING)'
+    # Match connection name + state: "CONN_NAME:  STATE"
+    STATUS_CONN_RE = r'^([^\s:]+):\s+(INSTALLED|CONNECTING|REKEYING|DELETING)'
+    STATUS_NOT_RUNNING_RE = (
+        r'No active Security Associations found\.?$')
+    # Match established SA lines like: "child_sa.*INSTALLED.*established"
+    STATUS_IPSEC_SA_ESTABLISHED_RE = (
+        r'(?:IKE|CHILD)_SA[^\n]*\n[^\n]*(?:ESTABLISHED|INSTALLED)')
+
+    binary = 'swanctl'
+    IPSEC_BINARY = 'ipsec'
+
+    def __init__(self, conf, process_id, vpnservice, namespace):
+        super().__init__(conf, process_id, vpnservice, namespace)
+        self._swanctl_config_dir = os.path.join(self.etc_dir, 'swanctl')
+        self._strongswan_piddir = self._get_strongswan_piddir()
+        self._rootwrap_cfg = self._get_rootwrap_config()
+        self.STATUS_CONN_PATTERN = re.compile(self.STATUS_CONN_RE)
+        LOG.debug("swanctl config dir is '%s'", self._swanctl_config_dir)
+
+    def _get_strongswan_piddir(self):
+        return utils.execute(
+            cmd=[self.IPSEC_BINARY, '--piddir'], run_as_root=True).strip()
+
+    def _execute(self, cmd, check_exit_code=True, extra_ok_codes=None):
+        """Execute command on namespace.
+
+        This execute is wrapped by namespace wrapper.
+        The namespace wrapper will bind /etc/ and /var/run
+        """
+        ip_wrapper = ip_lib.IPWrapper(namespace=self.namespace)
+        ns_wrapper = self.get_ns_wrapper()
+        return ip_wrapper.netns.execute(
+            [ns_wrapper,
+             '--mount_paths=/etc:{}/etc,{}:{}/var/run'.format(
+                 self.config_dir, self._strongswan_piddir, self.config_dir),
+             ('--rootwrap_config=%s' % self._rootwrap_cfg
+                 if self._rootwrap_cfg else ''),
+             '--cmd=%s' % ','.join(cmd)],
+            check_exit_code=check_exit_code,
+            extra_ok_codes=extra_ok_codes)
+
+    # ------------------------------------------------------------------ #
+    # Section 1: Configuration format
+    # ------------------------------------------------------------------ #
+
+    def _get_swanctl_connections_dir(self):
+        """Return path to swanctl connections directory."""
+        return os.path.join(
+            self._swanctl_config_dir, 'connections')
+
+    def _get_swanctl_secrets_dir(self):
+        """Return path to swanctl secrets directory."""
+        return os.path.join(
+            self._swanctl_config_dir, 'secrets')
+
+    def ensure_configs(self):
+        """Generate swanctl configuration files.
+
+        Creates the connections/ and secrets/ subdirectories under
+        the per-router etc/swanctl directory and renders the Jinja2
+        templates into those directories.
+        """
+        self.ensure_config_dir(self.vpnservice)
+        fileutils.ensure_tree(
+            self._get_swanctl_connections_dir(), 0o755)
+        fileutils.ensure_tree(self._get_swanctl_secrets_dir(), 0o755)
+
+        # Render swanctl.conf into connections/ directory
+        for ipsec_site_conn in (
+                self.vpnservice['ipsec_site_connections']):
+            conn_id = ipsec_site_conn['id']
+            config_str = self._gen_config_content(
+                cfg.CONF.strongswan.swanctl_config_template,
+                self.vpnservice)
+            # Write per-connection file for easier reload
+            conn_file = os.path.join(
+                self._get_swanctl_connections_dir(),
+                '%s.conf' % conn_id)
+            file_utils.replace_file(conn_file, config_str)
+
+        # Render secrets into secrets/ directory
+        secrets_str = self._gen_config_content(
+            cfg.CONF.strongswan.swanctl_secrets_template,
+            self.vpnservice)
+        secrets_file = os.path.join(self._get_swanctl_secrets_dir(),
+                                    'secrets.conf')
+        file_utils.replace_file(secrets_file, secrets_str, 0o600)
+
+    # ------------------------------------------------------------------ #
+    # Section 2: Daemon management (charon lifecycle in namespace)
+    # ------------------------------------------------------------------ #
+
+    def _start_daemon(self):
+        """Start strongSwan charon in the router namespace."""
+        self._execute([self.IPSEC_BINARY, 'start'],
+                      check_exit_code=False)
+
+    def _stop_daemon(self):
+        """Stop strongSwan charon in the router namespace."""
+        self._execute([self.IPSEC_BINARY, 'stop'],
+                      check_exit_code=False)
+
+    def _wait_for_charon_ready(self, timeout=10):
+        """Wait for charon to be ready by polling swanctl --list-sas.
+
+        :param timeout: Maximum seconds to wait.
+        :raises RuntimeError: If charon does not become ready within timeout.
+        """
+        start = time.time()
+        while time.time() - start < timeout:
+            try:
+                self._execute([self.binary, '--list-sas'],
+                              check_exit_code=False)
+                return
+            except RuntimeError:
+                time.sleep(0.5)
+        raise RuntimeError(
+            _('charon did not become ready within %ds' % timeout))
+
+    # ------------------------------------------------------------------ #
+    # Section 3: Connection control (stroke CLI → swanctl equivalents)
+    # ------------------------------------------------------------------ #
+
+    def _initiate_connection_swanctl(self, child_sa_id):
+        """Initiate an IPsec child SA via swanctl.
+
+        Equivalent to: ipsec stroke up-nb <id>
+        :param child_sa_id: The IPSec site connection ID.
+        """
+        self._execute([self.binary, '--initiate', '--child', child_sa_id])
+
+    def _reload_config_swanctl(self):
+        """Reload all swanctl configuration.
+
+        Equivalent to: ipsec reload
+        """
+        self._execute([self.binary, '--load-all'])
+
+    def _reload_secrets_swanctl(self):
+        """Reload credentials/secrets via swanctl.
+
+        Equivalent to: ipsec rereadsecrets
+        """
+        self._execute([self.binary, '--load-creds'])
+
+    # ------------------------------------------------------------------ #
+    # Section 4: Status parsing (swanctl --list-sas output format)
+    # ------------------------------------------------------------------ #
+
+    def _check_swanctl_status_line(self, line):
+        """Parse a swanctl --list-sas output line.
+
+        swanctl --list-sas produces indented blocks like::
+
+            VPN_abc123:  INSTALLED, ESTABLISHED, current (IPv4)
+                child_sa:  VPN_abc123_child_1
+                    INSTALLED, ESTABLISHED, current (IPv4)
+
+        This method extracts the connection/child SA name and state.
+
+        :param line: A single line from swanctl --list-sas output.
+        :returns: Tuple of (connection_id, status) or (None, None).
+        """
+        # Check for established CHILD_SA lines
+        m = self.STATUS_IPSEC_SA_ESTABLISHED_PATTERN.search(line)
+        if m:
+            # Extract the child_sa name from preceding context
+            return None, constants.ACTIVE
+
+        m = self.STATUS_PATTERN.search(line)
+        if m:
+            state_str = m.group(1)
+            status = self.STATUS_MAP.get(state_str, constants.DOWN)
+            # For lines like "child_sa:  NAME" followed by state on next line
+            return None, status
+
+        return None, None
+
+    def _extract_and_swanctl_connection_status(self, output):
+        """Parse swanctl --list-sas output and record connection statuses.
+
+        :param output: Full text output from swanctl --list-sas.
+        """
+        if not output:
+            self.connection_status = {}
+            return
+
+        # Parse the structured output line by line
+        lines = output.split('\n')
+        current_conn = None
+        for i, line in enumerate(lines):
+            # Match connection name at start of line: "CONN_NAME:  STATE"
+            conn_match = self.STATUS_CONN_PATTERN.match(line)
+            if conn_match:
+                current_conn = conn_match.group(1)
+                state = conn_match.group(2)
+                status = self.STATUS_MAP.get(state, constants.DOWN)
+                self._record_connection_status(current_conn, status)
+
+            # Match child_sa lines
+            child_match = re.match(
+                r'^\s+child_sa:\s+(\S+)', line)
+            if child_match and current_conn:
+                child_id = child_match.group(1)
+                # Check next line for state
+                if i + 1 < len(lines):
+                    next_line = lines[i + 1]
+                    state_match = self.STATUS_PATTERN.search(next_line)
+                    if state_match:
+                        state = state_match.group(1)
+                        status = self.STATUS_MAP.get(state, constants.DOWN)
+                        # Use child_sa name as connection_id for finer
+                        # granularity
+                        self._record_connection_status(child_id, status)
+
+    def get_status(self):
+        """Get status via swanctl --list-sas."""
+        return self._execute([self.binary, '--list-sas'],
+                             check_exit_code=False)
+
+    # ------------------------------------------------------------------ #
+    # Lifecycle methods (start/stop/reload)
+    # ------------------------------------------------------------------ #
+
+    def disable(self):
+        """Disable the swanctl process."""
+        try:
+            if self.active:
+                self.stop()
+            self.remove_config()
+        except RuntimeError:
+            LOG.exception(
+                "Failed to disable swanctl vpn process on router %s",
+                self.id)
+
+    def restart(self):
+        """Restart the swanctl process."""
+        self.reload_secrets()
+        self.reload()
+
+    def start(self):
+        """Start the swanctl process."""
+        if not self.namespace:
+            return
+        self._start_daemon()
+        self._wait_for_charon_ready()
+        self._reload_config_swanctl()
+        self._reload_secrets_swanctl()
+        for ipsec_site_conn in (
+                self.vpnservice['ipsec_site_connections']):
+            self._initiate_connection_swanctl(
+                ipsec_site_conn['id'])
+
+    def stop(self):
+        """Stop the swanctl process."""
+        self._stop_daemon()
+        self.connection_status = {}
+
+    def reload_secrets(self):
+        """Reload secrets via swanctl --load-creds."""
+        self._reload_secrets_swanctl()
+
+    def reload(self):
+        """Reload configuration via swanctl --load-all."""
+        self._reload_config_swanctl()
+
+    @property
+    def active(self):
+        """Check if the process is active or not."""
+        try:
+            status = self.get_status()
+            self._extract_and_swanctl_connection_status(status)
+            if not self.connection_status:
+                return False
+        except RuntimeError:
+            return False
+        return True
+
+
+class SwanctlDriver(ipsec.IPsecDriver):
+    """IPsec driver that uses SwanctlProcess.
+
+    .. versionadded:: 25.1.0
+    """
+
+    def create_process(self, process_id, vpnservice, namespace):
+        return SwanctlProcess(
+            self.conf,
+            process_id,
+            vpnservice,
+            namespace)
+
+
 class StrongSwanDriver(ipsec.IPsecDriver):
 
     def create_process(self, process_id, vpnservice, namespace):

neutron_vpnaas/services/vpn/device_drivers/template/swanctl/ipsec.secrets.template

@@ -0,0 +1,10 @@
+# swanctl secrets for {{vpnservice.id}}
+# Generated by neutron-vpnaas
+
+secrets {
+{% for ipsec_site_connection in vpnservice.ipsec_site_connections %}
+  - id: "{{ipsec_site_connection.local_id}}{{ipsec_site_connection.peer_id}}"
+    type: shared-secret
+    secret: "{{ipsec_site_connection.psk}}"
+{% endfor %}
+}

neutron_vpnaas/services/vpn/device_drivers/template/swanctl/swanctl.conf.template

@@ -0,0 +1,68 @@
+# swanctl configuration for {{vpnservice.id}}
+# Generated by neutron-vpnaas
+
+connections {
+{% for ipsec_site_connection in vpnservice.ipsec_site_connections %}
+  {{ipsec_site_connection.id}} {
+    local_addrs  = [ {{ipsec_site_connection.external_ip}} ]
+    remote_addrs = [ {{ipsec_site_connection.peer_address}} ]
+
+    peers {
+      local {
+        auth = psk
+        id   = {{ipsec_site_connection.local_id}}
+      }
+      remote {
+        auth = psk
+        id   = {{ipsec_site_connection.peer_id}}
+      }
+    }
+
+    ikesa {
+      version           = {{ipsec_site_connection.ikepolicy.ike_version}}
+      keylife            = {{ipsec_site_connection.ikepolicy.lifetime_value}}s
+      rekey_time         = 3m
+      rekey_margin       = 3m
+      keyingtries        = 1
+      mobike             = no
+      dpd_delay          = {{ipsec_site_connection.dpd_interval}}s
+      dpd_timeout        = {{ipsec_site_connection.dpd_timeout}}s
+      dpd_action         = {{ipsec_site_connection.dpd_action}}
+
+      proposals {
+        encryption = {{ipsec_site_connection.ikepolicy.encryption_algorithm}}
+        integrity  = {{ipsec_site_connection.ikepolicy.auth_algorithm}}
+        pfs        = {{ipsec_site_connection.ikepolicy.pfs}}
+      }
+    }
+
+    children {
+      {{ipsec_site_connection.id}}_child {
+        start_action = trap
+        req_mode     = tunnel
+        ipcomp         = no
+
+        local_ts  = [ {% for cidr in ipsec_site_connection['local_cidrs'] %}{{cidr}}{% if not loop.last %}, {% endif %}{% endfor %} ]
+        remote_ts = [ {% for cidr in ipsec_site_connection['peer_cidrs'] %}{{cidr}}{% if not loop.last %}, {% endif %}{% endfor %} ]
+
+        proposals {
+          encryption = {{ipsec_site_connection.ipsecpolicy.encryption_algorithm}}
+{% if ipsec_site_connection.ipsecpolicy.transform_protocol != "ah" and 'cm' not in ipsec_site_connection.ipsecpolicy.encryption_algorithm %}
+          integrity  = {{ipsec_site_connection.ipsecpolicy.auth_algorithm}}
+{% endif %}
+          pfs        = {{ipsec_site_connection.ipsecpolicy.pfs}}
+{% if ipsec_site_connection.ipsecpolicy.transform_protocol == "ah" %}
+          protocol   = ah
+{% else %}
+          protocol   = esp
+{% endif %}
+        }
+
+        lifetime {
+          seconds = {{ipsec_site_connection.ipsecpolicy.lifetime_value}}s
+        }
+      }
+    }
+  }
+{% endfor %}
+}

neutron_vpnaas/tests/functional/strongswan/test_ovn_swanctl.py

@@ -0,0 +1,20 @@
+# Copyright 2026 OpenStack Foundation
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+from neutron_vpnaas.tests.functional.common import ovn_base
+
+
+class TestOvnSwanctl(ovn_base.TestOvnVPNAgentBase):
+    VPN_DEVICE_DRIVER = ('neutron_vpnaas.services.vpn.device_drivers.'
+                         'ovn_ipsec.OvnSwanctlDriver')

neutron_vpnaas/tests/functional/strongswan/test_swanctl_driver.py

@@ -0,0 +1,70 @@
+# Copyright 2026 OpenStack Foundation
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import os
+import shutil
+import tempfile
+from unittest import mock
+
+from neutron.conf.agent.l3 import config as l3_config
+from neutron.tests.functional import base
+from oslo_config import cfg
+
+from neutron_vpnaas.services.vpn.device_drivers import ipsec
+from neutron_vpnaas.services.vpn.device_drivers import strongswan_ipsec
+from neutron_vpnaas.tests.unit.services.vpn.device_drivers import test_ipsec
+
+
+class TestSwanctlConfigGeneration(base.BaseSudoTestCase):
+
+    def setUp(self):
+        super().setUp()
+        self.conf = cfg.CONF
+        self.conf.register_opts(l3_config.OPTS)
+        self.conf.register_opts(ipsec.ipsec_opts, 'ipsec')
+        self.conf.register_opts(strongswan_ipsec.strongswan_opts,
+                                'strongswan')
+        self.tmpdir = tempfile.mkdtemp()
+        self.addCleanup(shutil.rmtree, self.tmpdir, ignore_errors=True)
+        self.conf.set_override('state_path', self.tmpdir)
+        self.process = strongswan_ipsec.SwanctlProcess(
+            self.conf,
+            'swanctl-test-router',
+            test_ipsec.FAKE_VPN_SERVICE,
+            mock.ANY)
+        self._swanctl_dir = os.path.join(
+            self.tmpdir, 'ipsec', 'swanctl-test-router', 'etc', 'swanctl')
+
+    def test_ensure_configs_writes_swanctl_files(self):
+        self.process.ensure_configs()
+        conn_dir = os.path.join(self._swanctl_dir, 'connections')
+        secrets_dir = os.path.join(self._swanctl_dir, 'secrets')
+        self.assertTrue(os.path.isdir(conn_dir))
+        self.assertTrue(os.path.isdir(secrets_dir))
+
+        conn_files = os.listdir(conn_dir)
+        self.assertEqual(2, len(conn_files))
+        for conn_file in conn_files:
+            with open(os.path.join(conn_dir, conn_file)) as fh:
+                content = fh.read()
+            self.assertIn('connections {', content)
+            self.assertIn('local_addrs', content)
+
+        secrets_file = os.path.join(secrets_dir, 'secrets.conf')
+        self.assertTrue(os.path.exists(secrets_file))
+        with open(secrets_file) as fh:
+            secrets_content = fh.read()
+        self.assertIn('secrets {', secrets_content)
+        self.assertIn('shared-secret', secrets_content)

neutron_vpnaas/tests/unit/services/vpn/device_drivers/test_ovn_ipsec.py

@@ -248,3 +248,14 @@ class TestOvnLibreSwanDriver(TestOvnStrongSwanDriver):
     def setUp(self):
         super().setUp(driver=ovn_ipsec.OvnLibreSwanDriver,
                       ipsec_process=libreswan_ipsec.LibreSwanProcess)
+
+
+class TestOvnSwanctlDriver(TestOvnStrongSwanDriver):
+    def setUp(self):
+        super().setUp(driver=ovn_ipsec.OvnSwanctlDriver,
+                      ipsec_process=strongswan_ipsec.SwanctlProcess)
+
+    def test_create_process(self):
+        process = self.driver.create_process(
+            FAKE_PROCESS_ID, self.vpnservice, FAKE_NAMESPACE_NAME)
+        self.assertIsInstance(process, strongswan_ipsec.SwanctlProcess)

neutron_vpnaas/tests/unit/services/vpn/device_drivers/test_strongswan_ipsec.py

@@ -0,0 +1,235 @@
+# Copyright 2026 OpenStack Foundation
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import copy
+import os
+import tempfile
+from unittest import mock
+import warnings
+
+from neutron_lib import constants
+from oslo_config import cfg
+
+from neutron_vpnaas.services.vpn.device_drivers import ipsec
+from neutron_vpnaas.services.vpn.device_drivers import strongswan_ipsec
+from neutron_vpnaas.tests import base
+from neutron_vpnaas.tests.unit.services.vpn.device_drivers import test_ipsec
+
+FAKE_CONN1_ID = test_ipsec.FAKE_IPSEC_SITE_CONNECTION1_ID
+FAKE_CONN2_ID = test_ipsec.FAKE_IPSEC_SITE_CONNECTION2_ID
+FAKE_PROCESS_ID = 'foo-process-id'
+
+SWANCTL_INSTALLED_STATUS = """%s:  INSTALLED, ESTABLISHED, current (IPv4)
+    child_sa:  %s_child
+        INSTALLED, ESTABLISHED, current (IPv4)
+""" % (FAKE_CONN2_ID, FAKE_CONN2_ID)
+
+SWANCTL_CONNECTING_STATUS = """%s:  CONNECTING, current (IPv4)
+""" % FAKE_CONN2_ID
+
+SWANCTL_EMPTY_STATUS = 'No active Security Associations found.'
+
+
+class TestSwanctlProcess(base.BaseTestCase):
+
+    def setUp(self):
+        super().setUp()
+        self.conf = cfg.CONF
+        self.conf.register_opts(ipsec.ipsec_opts, 'ipsec')
+        self.conf.register_opts(strongswan_ipsec.strongswan_opts, 'strongswan')
+        self.tmpdir = tempfile.mkdtemp()
+        self.conf.set_override('state_path', self.tmpdir)
+        self.vpnservice = copy.deepcopy(test_ipsec.FAKE_VPN_SERVICE)
+        self.process = strongswan_ipsec.SwanctlProcess(
+            self.conf,
+            FAKE_PROCESS_ID,
+            self.vpnservice,
+            'qrouter-test')
+        self._swanctl_dir = os.path.join(
+            self.tmpdir, 'ipsec', FAKE_PROCESS_ID, 'etc', 'swanctl')
+
+    def test_extract_connection_status_installed(self):
+        self.process._extract_and_swanctl_connection_status(
+            SWANCTL_INSTALLED_STATUS)
+        self.assertEqual(
+            constants.ACTIVE,
+            self.process.connection_status[FAKE_CONN2_ID]['status'])
+        self.assertEqual(
+            constants.ACTIVE,
+            self.process.connection_status[
+                '%s_child' % FAKE_CONN2_ID]['status'])
+
+    def test_extract_connection_status_connecting(self):
+        self.process._extract_and_swanctl_connection_status(
+            SWANCTL_CONNECTING_STATUS)
+        self.assertEqual(
+            constants.DOWN,
+            self.process.connection_status[FAKE_CONN2_ID]['status'])
+
+    def test_extract_connection_status_empty(self):
+        self.process.connection_status = {FAKE_CONN1_ID: {'status': 'foo'}}
+        self.process._extract_and_swanctl_connection_status('')
+        self.assertEqual({}, self.process.connection_status)
+
+    def test_check_swanctl_status_line(self):
+        conn_id, status = self.process._check_swanctl_status_line(
+            'INSTALLED, ESTABLISHED')
+        self.assertIsNone(conn_id)
+        self.assertEqual(constants.ACTIVE, status)
+
+    @mock.patch.object(strongswan_ipsec.SwanctlProcess,
+                       '_wait_for_charon_ready')
+    @mock.patch.object(strongswan_ipsec.SwanctlProcess, '_execute')
+    def test_start(self, mock_execute, _mock_wait):
+        self.process.start()
+        mock_execute.assert_has_calls([
+            mock.call(['ipsec', 'start'], check_exit_code=False),
+            mock.call(['swanctl', '--load-all']),
+            mock.call(['swanctl', '--load-creds']),
+            mock.call(['swanctl', '--initiate', '--child', FAKE_CONN1_ID]),
+            mock.call(['swanctl', '--initiate', '--child', FAKE_CONN2_ID]),
+        ], any_order=False)
+
+    def test_start_without_namespace(self):
+        self.process.namespace = None
+        with mock.patch.object(
+                strongswan_ipsec.SwanctlProcess, '_execute') as execute:
+            self.process.start()
+            execute.assert_not_called()
+
+    @mock.patch.object(strongswan_ipsec.SwanctlProcess, '_execute')
+    def test_stop(self, mock_execute):
+        self.process.stop()
+        mock_execute.assert_called_once_with(
+            ['ipsec', 'stop'], check_exit_code=False)
+        self.assertEqual({}, self.process.connection_status)
+
+    @mock.patch.object(strongswan_ipsec.SwanctlProcess, '_execute')
+    def test_reload(self, mock_execute):
+        self.process.reload()
+        mock_execute.assert_called_once_with(['swanctl', '--load-all'])
+
+    @mock.patch.object(strongswan_ipsec.SwanctlProcess, '_execute')
+    def test_reload_secrets(self, mock_execute):
+        self.process.reload_secrets()
+        mock_execute.assert_called_once_with(['swanctl', '--load-creds'])
+
+    @mock.patch.object(strongswan_ipsec.SwanctlProcess, '_execute')
+    def test_get_status(self, mock_execute):
+        mock_execute.return_value = SWANCTL_INSTALLED_STATUS
+        status = self.process.get_status()
+        self.assertEqual(SWANCTL_INSTALLED_STATUS, status)
+        mock_execute.assert_called_once_with(
+            ['swanctl', '--list-sas'], check_exit_code=False)
+
+    @mock.patch.object(strongswan_ipsec.SwanctlProcess, '_execute')
+    def test_active(self, mock_execute):
+        mock_execute.return_value = SWANCTL_INSTALLED_STATUS
+        self.assertTrue(self.process.active)
+
+    @mock.patch.object(strongswan_ipsec.SwanctlProcess, '_execute')
+    def test_active_empty_status(self, mock_execute):
+        mock_execute.return_value = SWANCTL_EMPTY_STATUS
+        self.assertFalse(self.process.active)
+
+    def test_ensure_configs(self):
+        self.process.ensure_configs()
+        conn_dir = os.path.join(self._swanctl_dir, 'connections')
+        secrets_dir = os.path.join(self._swanctl_dir, 'secrets')
+        self.assertTrue(os.path.isdir(conn_dir))
+        self.assertTrue(os.path.isdir(secrets_dir))
+        self.assertTrue(os.path.exists(
+            os.path.join(conn_dir, '%s.conf' % FAKE_CONN1_ID)))
+        self.assertTrue(os.path.exists(
+            os.path.join(conn_dir, '%s.conf' % FAKE_CONN2_ID)))
+        self.assertTrue(os.path.exists(
+            os.path.join(secrets_dir, 'secrets.conf')))
+
+    @mock.patch.object(strongswan_ipsec.SwanctlProcess,
+                       '_get_strongswan_piddir', return_value='/var/run')
+    def test_execute_uses_namespace_wrapper(self, _mock_piddir):
+        with mock.patch(
+                'neutron_vpnaas.services.vpn.device_drivers.strongswan_ipsec.'
+                'ip_lib.IPWrapper') as ip_wrapper_cls:
+            ip_wrapper = ip_wrapper_cls.return_value
+            self.process._execute(['swanctl', '--list-sas'])
+            ip_wrapper.netns.execute.assert_called_once()
+            cmd = ip_wrapper.netns.execute.call_args[0][0]
+            self.assertTrue(
+                any('neutron-vpn-netns-wrapper' in arg for arg in cmd))
+            self.assertIn('--cmd=swanctl,--list-sas', cmd[-1])
+
+    @mock.patch.object(strongswan_ipsec.StrongSwanProcess,
+                       '_get_strongswan_piddir', return_value='/var/run')
+    def test_strongswan_process_deprecation_warning(self, _mock_piddir):
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter('always')
+            strongswan_ipsec.StrongSwanProcess(
+                self.conf, 'id', self.vpnservice, mock.ANY)
+        self.assertTrue(any(
+            issubclass(w.category, DeprecationWarning) for w in caught))
+
+
+class TestSwanctlConfigGeneration(test_ipsec.BaseIPsecDeviceDriver):
+
+    def setUp(self):
+        super().setUp(
+            driver=strongswan_ipsec.SwanctlDriver,
+            ipsec_process=strongswan_ipsec.SwanctlProcess)
+        self.conf.register_opts(strongswan_ipsec.strongswan_opts, 'strongswan')
+        self.conf.set_override('state_path', '/tmp')
+        self.process = strongswan_ipsec.SwanctlProcess(
+            self.conf, 'foo-process-id', self.vpnservice, mock.ANY)
+        self.swanctl_template = self.conf.strongswan.swanctl_config_template
+        self.swanctl_secrets_template = (
+            self.conf.strongswan.swanctl_secrets_template)
+
+    def test_swanctl_config_template(self):
+        actual = self.process._gen_config_content(
+            self.swanctl_template, self.vpnservice)
+        self.assertIn('connections {', actual)
+        self.assertIn(FAKE_CONN1_ID, actual)
+        self.assertIn(FAKE_CONN2_ID, actual)
+        self.assertIn('local_addrs  = [ 60.0.0.4 ]', actual)
+        self.assertIn('remote_addrs = [ 60.0.0.5 ]', actual)
+        self.assertIn('%s_child' % FAKE_CONN1_ID, actual)
+
+    def test_swanctl_secrets_template(self):
+        actual = self.process._gen_config_content(
+            self.swanctl_secrets_template, self.vpnservice)
+        self.assertIn('secrets {', actual)
+        self.assertIn('type: shared-secret', actual)
+        self.assertIn('0scGFzc3dvcmQ=', actual)
+
+
+class TestSwanctlDriver(test_ipsec.IPsecStrongswanDeviceDriverLegacy):
+
+    def setUp(self):
+        super().setUp(
+            driver=strongswan_ipsec.SwanctlDriver,
+            ipsec_process=strongswan_ipsec.SwanctlProcess)
+
+    def test_create_process(self):
+        process = self.driver.create_process(
+            test_ipsec.FAKE_ROUTER_ID, self.vpnservice, 'qrouter-test')
+        self.assertIsInstance(process, strongswan_ipsec.SwanctlProcess)
+
+    def test_status_handling_for_active_connection(self):
+        self._test_status_handling_for_active_connection(
+            SWANCTL_INSTALLED_STATUS)
+
+    def test_status_handling_for_downed_connection(self):
+        self._test_status_handling_for_downed_connection(
+            SWANCTL_CONNECTING_STATUS)

releasenotes/notes/swanctl-support-fbb7fc6b94284f06.yaml

@@ -0,0 +1,20 @@
+---
+features:
+  - |
+    Added support for ``swanctl`` (VICI protocol) as an alternative to the
+    legacy stroke/starter interface in the strongSwan device driver. Operators
+    can now enable swanctl mode by setting ``use_swanctl = True`` under the
+    ``[strongswan]`` configuration section. This enables compatibility with
+    modern Linux distributions (Fedora, RHEL 9+, recent Ubuntu/Debian) that
+    ship strongSwan without the deprecated stroke plugin.
+
+deprecated:
+  - |
+    The ``use_swanctl`` configuration option under ``[strongswan]`` is now
+    deprecated and will be removed in the 2027.2 release. Operators should
+    migrate to using swanctl mode by default without requiring this option.
+
+  - |
+    The ``StrongSwanProcess`` class is now deprecated and will be removed in a
+    future release. Users should migrate to ``SwanctlProcess`` which is now the
+    default behavior when ``use_swanctl`` is not explicitly set to False.