Revert "feat: use Keystone auth for Neutron ML2->Undersync"

python/neutron-understack/neutron_understack/config.py

@@ -26,16 +26,6 @@
             "the '/etc/undersync/token' will be read instead."
         ),
     ),
-    cfg.BoolOpt(
-        "undersync_use_keystone_auth",
-        default=False,
-        help=(
-            "Use Keystone authentication for Undersync. "
-            "If True, the driver will use the existing Neutron service token "
-            "from the [keystone_authtoken] config section instead of the "
-            "static JWT token."
-        ),
-    ),
     cfg.BoolOpt(
         "undersync_dry_run", default=True, help="Call Undersync with dry-run mode"
     ),

python/neutron-understack/neutron_understack/neutron_understack_mech.py

@@ -1,8 +1,6 @@
 import logging
 from uuid import UUID
 
-from keystoneauth1 import loading as ks_loading
-from keystoneauth1 import session as ks_session
 from neutron_lib import constants as p_const
 from neutron_lib.api.definitions import portbindings
 from neutron_lib.callbacks import events
@@ -39,59 +37,11 @@ def connectivity(self):  # type: ignore
     def initialize(self):
         config.register_ml2_understack_opts(cfg.CONF)
         conf = cfg.CONF.ml2_understack
-
-        if conf.undersync_use_keystone_auth:
-            auth_token = self._get_keystone_service_token()
-            self.undersync = Undersync(
-                auth_token=auth_token,
-                api_url=conf.undersync_url,
-                use_keystone_auth=True,
-            )
-        else:
-            self.undersync = Undersync(conf.undersync_token, conf.undersync_url)
-
+        self.undersync = Undersync(conf.undersync_token, conf.undersync_url)
         self.ironic_client = IronicClient()
         self.trunk_driver = UnderStackTrunkDriver.create(self)
         self.subscribe()
 
-    def _get_keystone_service_token(self) -> str:
-        """Get a service token from Keystone using the Neutron service credentials.
-
-        This uses the existing [keystone_authtoken] configuration section
-        to obtain a token for authenticating with Undersync.
-
-        Returns:
-            str: The Keystone service token.
-
-        Raises:
-            Exception: If unable to obtain a token from Keystone.
-        """
-        try:
-            # Load credentials from the [keystone_authtoken] section
-            auth = ks_loading.load_auth_from_conf_options(
-                cfg.CONF, "keystone_authtoken"
-            )
-            # Create session manually to avoid missing config options
-            sess = ks_session.Session(auth=auth, timeout=30)
-
-            token = sess.get_token()
-            if not token:
-                raise ValueError("Obtained token is empty")
-
-            LOG.info(
-                "Successfully obtained Keystone service token for Undersync "
-                "authentication"
-            )
-            return token
-
-        except Exception as e:
-            LOG.error(
-                "Failed to obtain Keystone service token: %(error)s. "
-                "Please check your [keystone_authtoken] configuration.",
-                {"error": e},
-            )
-            raise
-
     def subscribe(self):
         registry.subscribe(
             routers.handle_router_interface_removal,
@@ -316,7 +266,7 @@ def _bind_port_segment(self, context: PortContext, segment):
         current_vlan_segment = utils.vlan_segment_for_physnet(context, vlan_group_name)
         if current_vlan_segment:
             LOG.info(
-                "vlan segment: %(segment)s already preset for physnet: %(physnet)s",
+                "vlan segment: %(segment)s already preset for physnet: " "%(physnet)s",
                 {"segment": current_vlan_segment, "physnet": vlan_group_name},
             )
             dynamic_segment = current_vlan_segment

python/neutron-understack/neutron_understack/tests/conftest.py

@@ -280,7 +280,7 @@ def ironic_client(mocker) -> IronicClient:
 @pytest.fixture
 def understack_driver(oslo_config, ironic_client) -> UnderstackDriver:
     driver = UnderstackDriver()
-    driver.undersync = Undersync("auth_token", "api_url", use_keystone_auth=False)
+    driver.undersync = Undersync("auth_token", "api_url")
     driver.ironic_client = ironic_client
     return driver
 

python/neutron-understack/neutron_understack/tests/test_neutron_understack_mech.py

@@ -1,9 +1,6 @@
 from dataclasses import dataclass
 
 import pytest
-from oslo_config import cfg
-
-from neutron_understack.neutron_understack_mech import UnderstackDriver
 
 
 class TestUpdatePortPostCommit:
@@ -98,59 +95,3 @@ class FakeContext:
             ]
 
         understack_driver.create_network_postcommit(FakeContext())
-
-
-class TestKeystoneAuthentication:
-    def test_initialize_with_keystone_auth(self, mocker, oslo_config):
-        """Test that driver initializes with Keystone authentication when enabled."""
-        oslo_config.config(
-            undersync_use_keystone_auth=True,
-            group="ml2_understack",
-        )
-
-        mock_auth = mocker.patch("keystoneauth1.loading.load_auth_from_conf_options")
-        mock_session_class = mocker.patch(
-            "neutron_understack.neutron_understack_mech.ks_session.Session"
-        )
-        mock_get_token = mocker.MagicMock(return_value="test_service_token")
-
-        mock_session_instance = mocker.MagicMock()
-        mock_session_instance.get_token = mock_get_token
-        mock_session_class.return_value = mock_session_instance
-
-        # Mock IronicClient to avoid config issues
-        mocker.patch("neutron_understack.neutron_understack_mech.IronicClient")
-
-        driver = UnderstackDriver()
-        driver.initialize()
-
-        mock_auth.assert_called_once_with(cfg.CONF, "keystone_authtoken")
-        mock_session_class.assert_called_once()
-        mock_get_token.assert_called_once()
-        assert driver.undersync.use_keystone_auth is True
-        assert driver.undersync.token == "test_service_token"
-
-    def test_initialize_with_jwt_auth(self, mocker, oslo_config):
-        """Test that driver initializes with JWT auth only."""
-        oslo_config.config(
-            undersync_use_keystone_auth=False,
-            undersync_token="test_jwt_token",
-            group="ml2_understack",
-        )
-
-        mock_auth = mocker.patch("keystoneauth1.loading.load_auth_from_conf_options")
-        mock_session = mocker.patch(
-            "keystoneauth1.loading.load_session_from_conf_options"
-        )
-
-        # Mock IronicClient to avoid config issues
-        mocker.patch("neutron_understack.neutron_understack_mech.IronicClient")
-
-        driver = UnderstackDriver()
-        driver.initialize()
-
-        # Should not call Keystone auth functions
-        mock_auth.assert_not_called()
-        mock_session.assert_not_called()
-        assert driver.undersync.use_keystone_auth is False
-        assert driver.undersync.token == "test_jwt_token"

python/neutron-understack/neutron_understack/tests/test_utils.py

@@ -635,55 +635,3 @@ def test_port_bound_to_uuid_when_agent_reports_hostname(self, mocker):
 
         assert result == "trunk-456"
         mock_ironic.baremetal_node_uuid.assert_called_once_with("gateway-host-1")
-
-
-class TestUndersyncAuthentication:
-    def test_undersync_with_keystone_auth(self, mocker):
-        """Test that Undersync client uses X-Auth-Token header when enabled."""
-        from neutron_understack.undersync import Undersync
-
-        client = Undersync(
-            auth_token="test_token",
-            api_url="http://test.api",
-            use_keystone_auth=True,
-        )
-
-        # Access the client property to trigger its creation
-        session = client.client
-
-        assert session.headers["Content-Type"] == "application/json"
-        assert session.headers["X-Auth-Token"] == "test_token"
-        assert "Authorization" not in session.headers
-
-    def test_undersync_with_jwt_auth(self, mocker):
-        """Test that Undersync client uses Authorization Bearer header with JWT."""
-        from neutron_understack.undersync import Undersync
-
-        client = Undersync(
-            auth_token="test_jwt_token",
-            api_url="http://test.api",
-            use_keystone_auth=False,
-        )
-
-        # Access the client property to trigger its creation
-        session = client.client
-
-        assert session.headers["Content-Type"] == "application/json"
-        assert session.headers["Authorization"] == "Bearer test_jwt_token"
-        assert "X-Auth-Token" not in session.headers
-
-    def test_undersync_default_to_jwt_auth(self, mocker):
-        """Test that Undersync client defaults to JWT auth when not specified."""
-        from neutron_understack.undersync import Undersync
-
-        client = Undersync(
-            auth_token="test_token",
-            api_url="http://test.api",
-        )
-
-        # Access the client property to trigger its creation
-        session = client.client
-
-        assert session.headers["Content-Type"] == "application/json"
-        assert session.headers["Authorization"] == "Bearer test_token"
-        assert "X-Auth-Token" not in session.headers

python/neutron-understack/neutron_understack/undersync.py

@@ -19,26 +19,12 @@ def __init__(
         auth_token: str | None = None,
         api_url: str | None = None,
         timeout: int = 90,
-        use_keystone_auth: bool = False,
     ) -> None:
-        """Simple client for Undersync.
-
-        Args:
-            auth_token: Authentication token. If use_keystone_auth is True,
-                       this should be a Keystone service token. Otherwise,
-                       it should be a JWT token. If not provided, it will be
-                       fetched from /etc/undersync/token.
-            api_url: Undersync API URL.
-            timeout: Request timeout in seconds.
-            use_keystone_auth: If True, use X-Auth-Token header for Keystone
-                             authentication. Otherwise, use Authorization:
-                             Bearer header for JWT authentication.
-        """
+        """Simple client for Undersync."""
         self.token = auth_token or self._fetch_undersync_token()
         self.url = "http://undersync.undersync.svc.cluster.local:8080"
         self.api_url = api_url or self.url
         self.timeout = timeout
-        self.use_keystone_auth = use_keystone_auth
 
     def _fetch_undersync_token(self) -> str:
         file = pathlib.Path("/etc/undersync/token")
@@ -65,13 +51,10 @@ def sync_devices(
     @cached_property
     def client(self):
         session = requests.Session()
-        session.headers = {"Content-Type": "application/json"}
-
-        if self.use_keystone_auth:
-            session.headers["X-Auth-Token"] = self.token
-        else:
-            session.headers["Authorization"] = f"Bearer {self.token}"
-
+        session.headers = {
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {self.token}",
+        }
         return session
 
     def _undersync_post(self, action: str, vlan_group: str) -> requests.Response:

python/neutron-understack/pyproject.toml

@@ -66,6 +66,4 @@ target-version = "py312"
     "S311",     # allow non-cryptographic secure bits for test data
     "S101",
     "RUF012",   # test fixture classes often use mutable class-level data
-    "S105",     # allow hardcoded passwords for testing
-    "S106",     # allow hardcoded passwords for testing
 ]