feat(hyperv): implement kerberos session auth

.packit.yaml

@@ -12,3 +12,11 @@ files_to_sync:
 upstream_package_name: virt-who
 # downstream (Fedora) RPM package name
 downstream_package_name: virt-who
+
+jobs:
+  - job: copr_build
+    trigger: pull_request
+    targets:
+      - centos-stream-10
+      - fedora-all
+      - rhel-10-x86_64

scripts/container-pre-test.sh

@@ -1,7 +1,7 @@
 #!/bin/bash
 
 dnf install -y libnl3-devel python3-libvirt python3-dateutil python3-setuptools python3-pip dnf-plugins-core \
-    python3-requests python3-cryptography python3-subscription-manager-rhsm subscription-manager
+    python3-requests python3-cryptography python3-requests-gssapi python3-subscription-manager-rhsm subscription-manager
 dnf builddep -y virt-who.spec
 
 pip install -r requirements.txt

tests/test_hyperv.py

@@ -31,7 +31,7 @@
 from proxy import Proxy
 
 from virtwho import DefaultInterval
-from virtwho.virt.hyperv.hyperv import HyperV, HypervConfigSection
+from virtwho.virt.hyperv.hyperv import HyperV, HypervConfigSection, HyperVException
 from virtwho.virt import VirtError, Guest, Hypervisor, StatusReport
 
 
@@ -334,3 +334,141 @@ def test_proxy_if_html_parse_error_only_status_code_is_logged(self, logger_debug
         logger_debug.assert_called_with(
             'Invalid response (403) from Hyper-V (error: not well-formed (invalid token): line 8, column 18)'
         )
+
+
+class TestHyperVKerberosConnect(TestBase):
+    """Tests for the kerberos auth wiring in HyperV.connect()."""
+
+    def _make_hyperv(self, auth_method='kerberos', keytab=None, principal=None):
+        config_values = {
+            'type': 'hyperv',
+            'server': 'hyperv.example.com',
+            'auth_method': auth_method,
+            'owner': 'owner',
+        }
+        if keytab:
+            config_values['kerberos_keytab'] = keytab
+        if principal:
+            config_values['kerberos_principal'] = principal
+        config = HypervConfigSection('test', None)
+        config.update(**config_values)
+        config.validate()
+        return HyperV(self.logger, config, None, interval=DefaultInterval)
+
+    @patch('requests.Session')
+    def test_basic_auth_uses_hypervauth(self, session):
+        """auth_method=basic uses the HyperVAuth handler."""
+        from virtwho.virt.hyperv.hyperv import HyperVAuth
+        config_values = {
+            'type': 'hyperv',
+            'server': 'hyperv.example.com',
+            'auth_method': 'basic',
+            'username': 'user',
+            'password': 'pass',
+            'owner': 'owner',
+        }
+        config = HypervConfigSection('test', None)
+        config.update(**config_values)
+        config.validate()
+        hyperv = HyperV(self.logger, config, None, interval=DefaultInterval)
+
+        s = hyperv.connect()
+        self.assertIsInstance(s.auth, HyperVAuth)
+
+    @patch('virtwho.virt.hyperv.hyperv.gssapi.Credentials')
+    @patch('requests.Session')
+    def test_kerberos_auth_uses_httpspnegoauth(self, session, mock_credentials):
+        """auth_method=kerberos uses HTTPSPNEGOAuth."""
+        from requests_gssapi import HTTPSPNEGOAuth
+        hyperv = self._make_hyperv()
+        s = hyperv.connect()
+        self.assertIsInstance(s.auth, HTTPSPNEGOAuth)
+
+    @patch('virtwho.virt.hyperv.hyperv.gssapi.Credentials')
+    @patch('requests.Session')
+    def test_kerberos_sets_principal(self, session, mock_credentials):
+        """kerberos_principal is used to acquire gssapi.Credentials passed to HTTPSPNEGOAuth."""
+        sentinel_creds = object()
+        mock_credentials.return_value = sentinel_creds
+
+        hyperv = self._make_hyperv(principal='virtwho@EXAMPLE.COM')
+        s = hyperv.connect()
+
+        called_name = mock_credentials.call_args.kwargs['name']
+        self.assertEqual(str(called_name), 'virtwho@EXAMPLE.COM')
+        self.assertEqual(mock_credentials.call_args.kwargs['usage'], 'initiate')
+        self.assertIs(s.auth.creds, sentinel_creds)
+
+    @patch('virtwho.virt.hyperv.hyperv.gssapi.Credentials')
+    @patch('requests.Session')
+    def test_kerberos_principal_credential_failure_raises_clean_error(self, session, mock_credentials):
+        """A GSSError while acquiring credentials for the principal surfaces as HyperVException."""
+        import gssapi
+
+        class FakeGSSError(gssapi.exceptions.GSSError):
+            def __str__(self):
+                return 'no matching credentials in cache collection'
+
+        error = FakeGSSError.__new__(FakeGSSError)
+        BaseException.__init__(error, 'no matching credentials in cache collection')
+        mock_credentials.side_effect = error
+
+        hyperv = self._make_hyperv(principal='virtwho@EXAMPLE.COM')
+        self.assertRaises(HyperVException, hyperv.connect)
+
+    @patch('virtwho.virt.hyperv.hyperv.gssapi.Credentials')
+    @patch('requests.Session')
+    def test_kerberos_sets_keytab_store(self, session, mock_credentials):
+        """kerberos_keytab is passed to gssapi.Credentials via the 'store' argument."""
+        import tempfile
+        with tempfile.NamedTemporaryFile(delete=False) as f:
+            keytab_path = f.name
+        try:
+            hyperv = self._make_hyperv(keytab=keytab_path)
+            hyperv.connect()
+
+            called_store = mock_credentials.call_args.kwargs['store']
+            self.assertEqual(called_store, {"client_keytab": "FILE:%s" % keytab_path})
+        finally:
+            os.unlink(keytab_path)
+
+    @patch('virtwho.virt.hyperv.hyperv.gssapi.Credentials')
+    @patch('requests.Session')
+    def test_kerberos_no_principal_or_keytab_uses_defaults(self, session, mock_credentials):
+        """Without kerberos_principal/kerberos_keytab, gssapi.Credentials is still called,
+        with name and store left as None so the default credential cache is used."""
+        sentinel_creds = Mock(name='resolved-creds')
+        sentinel_creds.name = 'defaulted-user@EXAMPLE.COM'
+        mock_credentials.return_value = sentinel_creds
+
+        hyperv = self._make_hyperv()
+        s = hyperv.connect()
+
+        mock_credentials.assert_called_once_with(name=None, store=None, usage='initiate')
+        self.assertIs(s.auth.creds, sentinel_creds)
+
+    @patch('virtwho.virt.hyperv.hyperv.gssapi.Credentials')
+    @patch('requests.Session')
+    def test_kerberos_connect_runs_oneshot(self, session, mock_credentials):
+        """Kerberos auth path can complete a full oneshot run with mocked responses."""
+        session.return_value.post.side_effect = HyperVMock.post
+        hyperv = self._make_hyperv()
+        hyperv._oneshot = True
+        hyperv.dest = Queue()
+        hyperv._terminate_event = Event()
+        hyperv._interval = 0
+        hyperv._run()
+
+        session.return_value.post.assert_called()
+
+    @patch('virtwho.virt.hyperv.hyperv.gssapi.Credentials')
+    @patch('requests.Session')
+    def test_kerberos_401_raises_auth_failed(self, session, mock_credentials):
+        """A 401 after kerberos negotiate raises HyperVAuthFailed."""
+        session.return_value.post.return_value.status_code = 401
+        hyperv = self._make_hyperv()
+        hyperv._oneshot = True
+        hyperv.dest = Queue()
+        hyperv._terminate_event = Event()
+        hyperv._interval = 0
+        self.assertRaises(VirtError, hyperv._run)

virt-who.spec

@@ -19,6 +19,8 @@ Requires:       python3-libvirt
 Requires:       python3-subscription-manager-rhsm > 1.25.6
 Requires:       python3-cryptography
 Requires:       python3-requests
+# requests-gssapi required for Hyper-V Kerberos authentication support
+Requires:       python3-requests-gssapi
 Requires:       python3-pyyaml
 
 Requires: python3-systemd

virtwho/virt/hyperv/hyperv.py

@@ -25,6 +25,9 @@
 from xml.etree import ElementTree
 from requests.auth import AuthBase
 import requests
+import gssapi
+from requests_gssapi import HTTPSPNEGOAuth, OPTIONAL
+from requests_gssapi.exceptions import SPNEGOExchangeError, MutualAuthenticationError
 
 from virtwho import virt
 from virtwho.config import VirtConfigSection, accessible_file
@@ -353,6 +356,12 @@ def post(self, body):
         }
         try:
             response = self.connection.post(self.url, body, headers=headers)
+        except (SPNEGOExchangeError, MutualAuthenticationError) as e:
+            raise HyperVAuthFailed(
+                "Kerberos authentication failed: %s. "
+                "Verify that the keytab and principal are correct, "
+                "and that the KDC is reachable." % str(e)
+            )
         except requests.RequestException as e:
             raise HyperVException("Unable to connect to Hyper-V server: %s" % str(e))
 
@@ -492,8 +501,11 @@ def __init__(self, logger, config, dest, terminate_event=None,
                                      oneshot=oneshot,
                                      status=status)
         self.url = self.config['url']
-        self.username = self.config['username']
-        self.password = self.config['password']
+        self.auth_method = self.config.get('auth_method', 'basic')
+        self.username = self.config.get('username', '')
+        self.password = self.config.get('password', '')
+        self.kerberos_keytab = self.config.get('kerberos_keytab', None)
+        self.kerberos_principal = self.config.get('kerberos_principal', None)
 
         # First try to use old API (root/virtualization namespace) if doesn't
         # work, go with root/virtualization/v2
@@ -504,7 +516,34 @@ def connect(self):
         s = requests.Session()
         adapter = requests.adapters.HTTPAdapter(pool_connections=1, pool_maxsize=1)
         s.mount('http://', adapter)
-        s.auth = HyperVAuth(self.username, self.password, self.logger)
+
+        if self.auth_method == 'kerberos':
+            self.logger.debug('Using Kerberos (SPNEGO/Negotiate) authentication')
+            name = None
+            store = None
+
+            if self.kerberos_keytab:
+                store = {"client_keytab": f"FILE:{self.kerberos_keytab}"}
+                self.logger.debug('Using Kerberos keytab: %s', self.kerberos_keytab)
+
+            if self.kerberos_principal:
+                name = gssapi.Name(self.kerberos_principal, name_type=gssapi.NameType.kerberos_principal)
+                self.logger.debug('Using Kerberos principal: %s', self.kerberos_principal)
+
+            try:
+                # If 'name' is omitted, python-gssapi pulls the default principal from the 'store' (keytab)
+                # or the default credential cache. If 'store' is omitted, it looks in the default cache.
+                creds = gssapi.Credentials(name=name, store=store, usage="initiate")
+            except gssapi.exceptions.GSSError as e:
+                raise HyperVException(f"Failed to acquire Kerberos credentials: {e}")
+
+            if name is None:
+                self.logger.debug('Resolved default Kerberos principal: %s', creds.name)
+
+            s.auth = HTTPSPNEGOAuth(creds=creds, mutual_authentication=OPTIONAL)
+        else:
+            self.logger.debug('Using Basic authentication')
+            s.auth = HyperVAuth(self.username, self.password, self.logger)
         return s
 
     @classmethod