feat: add optional SW360KeycloakClient with automatic token renewal

ChangeLog.md

@@ -7,6 +7,10 @@
 
 ## NEXT
 
+* New optional client `SW360KeycloakClient` which takes a Keycloak `client_id` and
+  `client_secret` instead of a static token and renews the token shortly before it
+  expires. This is meant for long running operations where the token may expire
+  mid-run. `SW360` and its static token usage are unchanged. See Readme.md.
 * Fix unintentional parallel changes in `api_get_all()` which broke the `SW360Response` feature
   in V1.12.0.dev2. See Readme.md for details how the `SW360Response` class can now be used.
 * Dependency updates, especially because of PyJWT (CVE-2026-102272, CVE-2026-102273).

Readme.md

@@ -55,6 +55,22 @@ Install sw360 and required dependencies:
   r = client.get_release(release_id)
   ```
 
+* For long running operations the token may expire while the operation is still
+  running. In that case you can use `SW360KeycloakClient`, which takes the Keycloak
+  credentials instead of a token and renews the token shortly before it expires
+  (five minutes by default, see `refresh_before_seconds`):
+
+  ```python
+  from sw360 import SW360KeycloakClient
+  client = SW360KeycloakClient(sw360_url, client_id, client_secret, write_access=False)
+  client.login_api()
+  r = client.get_release(release_id)
+  ```
+
+  The client uses its own session, so it does not affect clients created with a
+  static token. If you need to hand the token to another component which creates
+  its own client, use `client.current_token` to get a token that is valid now.
+
 * Starting with v1.12, most responses are wrapped in `SW360Response`, a `dict` subclass,
   so existing code keeps working unchanged, but you get new convenience methods for the
   `_links` and `_embedded` sections of the SW360 HAL responses:

sw360/__init__.py

@@ -12,13 +12,14 @@
 from .base import SW360Response
 from .sorting import (BaseSortMixin, ProjectSortColumn, ReleaseSortColumn,
                       SortParam)
-from .sw360_api import SW360
+from .sw360_api import SW360, SW360KeycloakClient
 from .sw360error import SW360Error
 from .sw360keycloak import SW360Keycloak
 from .sw360oauth2 import SW360OAuth2
 
 __all__ = [
     "SW360",
+    "SW360KeycloakClient",
     "SW360Error",
     "SW360OAuth2",
     "SW360Keycloak",

sw360/sw360_api.py

@@ -10,10 +10,12 @@
 
 """Python interface to the Siemens SW360 platform"""
 
-from typing import Any, Dict, Optional, Tuple
+from typing import Any, Callable, Dict, Optional, Tuple
 
 import requests
 from requests.adapters import HTTPAdapter
+from requests.auth import AuthBase
+from requests.models import PreparedRequest
 from urllib3.util.retry import Retry
 
 from .attachments import AttachmentsMixin
@@ -26,6 +28,7 @@
 from .releases import ReleasesMixin
 from .reports import ReportsMixin
 from .sw360error import SW360Error
+from .sw360keycloak import KeycloakTokenProvider, SW360Keycloak
 from .vendor import VendorMixin
 from .vulnerabilities import VulnerabilitiesMixin
 
@@ -42,6 +45,18 @@
 session_default.mount("https://", adapter)
 
 
+class _TokenProviderAuth(AuthBase):
+    """Sets the authorization header of each request from a token provider."""
+
+    def __init__(self, token_provider: Callable[[], str], scheme: str) -> None:
+        self.token_provider = token_provider
+        self.scheme = scheme
+
+    def __call__(self, request: PreparedRequest) -> PreparedRequest:
+        request.headers["Authorization"] = f"{self.scheme} {self.token_provider()}"
+        return request
+
+
 class SW360(
     AttachmentsMixin,
     ClearingMixin,
@@ -79,10 +94,10 @@ class SW360(
     def __init__(
         self,
         url: str,
-        token: str,
+        token: str = "",
         oauth2: bool = False,
         session: Optional[requests.Session] = session_default,
-        default_batch_size: int = 50
+        default_batch_size: int = 50,
     ) -> None:
         """Constructor"""
         super().__init__(url, token, oauth2, default_batch_size)
@@ -198,3 +213,92 @@ def parse_version(version_string: str) -> Tuple[int, int]:
         """Extract major.minor as comparable tuple"""
         parts = version_string.split('.')
         return (int(parts[0]), int(parts[1]))
+
+
+class SW360KeycloakClient(SW360):
+    """Python interface to the Siemens SW360 platform, authenticating via Keycloak.
+
+    In contrast to `SW360`, this client is given the Keycloak credentials instead
+    of a token. It obtains the token itself and renews it shortly before it
+    expires, so that operations running longer than the token lifetime do not
+    fail. `SW360` and its static token usage are not affected by this class.
+
+    Unless a session is passed in, the client creates its own session, so that
+    the session shared by `SW360` instances is left unchanged. A session passed
+    in by the caller will have its authentication configured by this client.
+
+    :param url: URL of the SW360 instance
+    :param client_id: the SW360 client_id to be used for token generation
+    :param client_secret: the SW360 client_secret to be used for token generation
+    :param write_access: whether the token should have write access
+    :param session: the session to use, a new one is created if not given
+    :param refresh_before_seconds: how long before expiry a token gets renewed
+    :type url: string
+    :type client_id: string
+    :type client_secret: string
+    :type write_access: boolean
+    :type session: requests.Session
+    :type refresh_before_seconds: int
+    :raises SW360Error: if no token could be obtained from Keycloak
+    """
+
+    def __init__(
+        self,
+        url: str,
+        client_id: str,
+        client_secret: str,
+        write_access: bool = False,
+        session: Optional[requests.Session] = None,
+        default_batch_size: int = 50,
+        refresh_before_seconds: int = 300,
+    ) -> None:
+        """Constructor"""
+        self._keycloak_token_provider = KeycloakTokenProvider(
+            SW360Keycloak(url),
+            client_id,
+            client_secret,
+            write_access=write_access,
+            refresh_before_seconds=refresh_before_seconds,
+        )
+        initial_token = self._keycloak_token_provider()
+
+        if session is None:
+            session = requests.Session()
+            session.mount("http://", adapter)
+            session.mount("https://", adapter)
+
+        super().__init__(
+            url=url,
+            token=initial_token,
+            oauth2=True,
+            session=session,
+            default_batch_size=default_batch_size,
+        )
+        if self.session is not None:
+            self.session.auth = _TokenProviderAuth(self._keycloak_token_provider, "Bearer")
+
+    @property
+    def current_token(self) -> str:
+        """The access token currently in use, renewed first if it is about to
+        expire. Use this if you need to pass the token on, for example to
+        another component creating its own client.
+
+        :return: a currently valid access token
+        :rtype: str
+        :raises SW360Error: if no token could be obtained from Keycloak
+        """
+        return self._keycloak_token_provider()
+
+    @property
+    def api_headers(self) -> Dict[str, str]:
+        """The HTTP headers to use for requests that do not go through the
+        session, with a token that is valid now."""
+        headers = self._api_headers.copy()
+        headers["Authorization"] = "Bearer " + self._keycloak_token_provider()
+        return headers
+
+    @api_headers.setter
+    def api_headers(self, headers: Dict[str, str]) -> None:
+        # the base constructor sets the initial static header, the getter
+        # replaces its token on every read
+        self._api_headers = headers

sw360/sw360keycloak.py

@@ -7,7 +7,7 @@
 # SPDX-License-Identifier: MIT
 # -------------------------------------------------------------------------------
 
-from datetime import datetime
+from datetime import datetime, timedelta
 from typing import Optional
 
 import jwt
@@ -136,3 +136,84 @@ def is_write_token(self, token: str) -> bool:
             return scope.lower().find("write") >= 0
         else:
             raise SW360Error(message="Unable to analyze token: scope field missing!")
+
+
+class KeycloakTokenProvider:
+    """Provides a valid Keycloak access token to an SW360 client.
+
+    The token is obtained using the client credentials grant and renewed shortly
+    before it expires, so that long running operations do not fail with an
+    expired token. A provider belongs to a single client; it must not be shared
+    between clients using different credentials or scopes.
+
+    :param keycloak: the Keycloak helper used to request tokens
+    :param client_id: the SW360 client_id to be used for token generation
+    :param client_secret: the SW360 client_secret to be used for token generation
+    :param write_access: whether the token should have write access
+    :param refresh_before_seconds: how long before expiry a token gets renewed
+    :type keycloak: SW360Keycloak
+    :type client_id: str
+    :type client_secret: str
+    :type write_access: bool
+    :type refresh_before_seconds: int
+    """
+
+    def __init__(
+        self,
+        keycloak: SW360Keycloak,
+        client_id: str,
+        client_secret: str,
+        write_access: bool = False,
+        refresh_before_seconds: int = 300,
+    ) -> None:
+        self.keycloak = keycloak
+        self.client_id = client_id
+        self.client_secret = client_secret
+        self.write_access = write_access
+        self.refresh_before = timedelta(seconds=refresh_before_seconds)
+        self._token = ""
+
+    def __call__(self) -> str:
+        """
+        Returns a token that is valid now, renewing the cached one if it is
+        about to expire.
+
+        :return: a currently valid access token
+        :rtype: str
+        :raises SW360Error: if no token could be obtained from Keycloak
+        """
+
+        if not self._token or self._expires_soon(self._token):
+            token = self.keycloak.get_keycloak_token(
+                self.client_id,
+                self.client_secret,
+                write_access=self.write_access,
+            )
+            if not token:
+                raise SW360Error(message="Unable to obtain token from Keycloak")
+            self._token = token
+
+        return self._token
+
+    def _expires_soon(self, token: str) -> bool:
+        """
+        Checks if the given JWT token expires within the renewal buffer.
+
+        :param token: the JWT token to check
+        :type token: str
+        :return: True if the token should be renewed, False otherwise
+        :rtype: bool
+        :raises SW360Error: if the token cannot be analyzed
+        """
+
+        try:
+            # alg = RS256
+            decoded = jwt.decode(token, algorithms=["HS256"], options={"verify_signature": False})
+        except Exception as ex:
+            raise SW360Error(message="Unable to analyze token: " + repr(ex))
+
+        if "exp" not in decoded:
+            raise SW360Error(message="Unable to analyze token: exp field missing!")
+
+        expiry = datetime.fromtimestamp(int(decoded["exp"]))
+        return expiry <= datetime.now() + self.refresh_before