From 2cb566ee249ec0baf01ad84bf3cb3344e93ec95d Mon Sep 17 00:00:00 2001 From: Berke Yoncaci Date: Wed, 30 Sep 2026 17:01:52 +0300 Subject: [PATCH] feat: add optional SW360KeycloakClient with automatic token renewal SW360 is given a finished token and cannot renew it, so operations running longer than the token lifetime fail once it expires. SW360KeycloakClient takes the Keycloak client_id and client_secret instead and renews the token shortly before expiry. SW360 and SW360Keycloak are unchanged, existing callers keep working as before. --- ChangeLog.md | 4 + Readme.md | 16 ++++ sw360/__init__.py | 3 +- sw360/sw360_api.py | 110 ++++++++++++++++++++++++- sw360/sw360keycloak.py | 83 ++++++++++++++++++- tests/test_sw360_base.py | 171 ++++++++++++++++++++++++++++++++++++++- 6 files changed, 381 insertions(+), 6 deletions(-) diff --git a/ChangeLog.md b/ChangeLog.md index 849ba71..b83d7d7 100644 --- a/ChangeLog.md +++ b/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). diff --git a/Readme.md b/Readme.md index 948c263..354a1ff 100644 --- a/Readme.md +++ b/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: diff --git a/sw360/__init__.py b/sw360/__init__.py index f0847c3..09f41fa 100644 --- a/sw360/__init__.py +++ b/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", diff --git a/sw360/sw360_api.py b/sw360/sw360_api.py index 8516689..cca8e93 100644 --- a/sw360/sw360_api.py +++ b/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 diff --git a/sw360/sw360keycloak.py b/sw360/sw360keycloak.py index 03f14e9..2ca3dc3 100644 --- a/sw360/sw360keycloak.py +++ b/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 diff --git a/tests/test_sw360_base.py b/tests/test_sw360_base.py index 50dbe2e..88d260c 100644 --- a/tests/test_sw360_base.py +++ b/tests/test_sw360_base.py @@ -10,16 +10,21 @@ import json import os import sys +import time import unittest +from unittest.mock import patch +import jwt +import requests import responses from sw360.base import BaseMixin +from sw360.sw360_api import session_default sys.path.insert(1, "..") -from sw360 import SW360, SW360Error # noqa: E402 +from sw360 import SW360, SW360Error, SW360KeycloakClient # noqa: E402 class Sw360Test(unittest.TestCase): @@ -44,6 +49,170 @@ def test_constructor(self) -> None: lib = SW360("https://my.server.com", self.MYTOKEN, False) self.assertEqual(self.MYURL, lib.url) + def _make_token(self, valid_for_seconds: int, scope: str = "READ profile email") -> str: + signing_key = "test-signing-key-that-is-at-least-32-bytes" + return jwt.encode( + {"exp": int(time.time()) + valid_for_seconds, "scope": scope}, + signing_key, algorithm="HS256") + + def test_keycloak_client_uses_credentials_instead_of_token(self) -> None: + token = self._make_token(600) + + with patch("sw360.sw360_api.SW360Keycloak") as keycloak_class: + keycloak = keycloak_class.return_value + keycloak.get_keycloak_token.return_value = token + lib = SW360KeycloakClient(self.MYURL, "client-id", "client-secret") + + self.assertIsInstance(lib, SW360) + self.assertEqual(self.MYURL, lib.url) + self.assertEqual("Bearer " + token, lib.api_headers["Authorization"]) + keycloak.get_keycloak_token.assert_called_once_with( + "client-id", "client-secret", write_access=False) + + def test_keycloak_client_renews_token_before_expiry(self) -> None: + initial_token = self._make_token(600) + expiring_token = self._make_token(30) + fresh_token = self._make_token(600) + + with patch("sw360.sw360_api.SW360Keycloak") as keycloak_class: + keycloak = keycloak_class.return_value + keycloak.get_keycloak_token.side_effect = [initial_token, fresh_token] + + lib = SW360KeycloakClient(self.MYURL, "client-id", "client-secret") + self.assertEqual("Bearer " + initial_token, lib.api_headers["Authorization"]) + + # simulate that the cached token is now close to expiry + lib._keycloak_token_provider._token = expiring_token + self.assertEqual("Bearer " + fresh_token, lib.api_headers["Authorization"]) + self.assertEqual(2, keycloak.get_keycloak_token.call_count) + + def test_keycloak_client_keeps_token_that_is_still_valid(self) -> None: + token = self._make_token(600) + + with patch("sw360.sw360_api.SW360Keycloak") as keycloak_class: + keycloak = keycloak_class.return_value + keycloak.get_keycloak_token.return_value = token + + lib = SW360KeycloakClient(self.MYURL, "client-id", "client-secret") + for _ in range(3): + self.assertEqual("Bearer " + token, lib.api_headers["Authorization"]) + + self.assertEqual(1, keycloak.get_keycloak_token.call_count) + + def test_keycloak_client_honours_refresh_buffer(self) -> None: + initial_token = self._make_token(600) + fresh_token = self._make_token(1200) + + with patch("sw360.sw360_api.SW360Keycloak") as keycloak_class: + keycloak = keycloak_class.return_value + keycloak.get_keycloak_token.side_effect = [initial_token, fresh_token] + + # a token valid for 600s is renewed when the buffer is larger + lib = SW360KeycloakClient( + self.MYURL, "client-id", "client-secret", refresh_before_seconds=900) + self.assertEqual("Bearer " + fresh_token, lib.api_headers["Authorization"]) + + def test_keycloak_client_requests_write_access(self) -> None: + token = self._make_token(600, scope="READ WRITE profile email") + + with patch("sw360.sw360_api.SW360Keycloak") as keycloak_class: + keycloak = keycloak_class.return_value + keycloak.get_keycloak_token.return_value = token + SW360KeycloakClient(self.MYURL, "client-id", "client-secret", write_access=True) + + keycloak.get_keycloak_token.assert_called_once_with( + "client-id", "client-secret", write_access=True) + + def test_keycloak_client_reports_missing_token(self) -> None: + with patch("sw360.sw360_api.SW360Keycloak") as keycloak_class: + keycloak_class.return_value.get_keycloak_token.return_value = None + with self.assertRaises(SW360Error) as context: + SW360KeycloakClient(self.MYURL, "client-id", "client-secret") + + self.assertEqual("Unable to obtain token from Keycloak", context.exception.message) + + def test_keycloak_client_reports_token_that_cannot_be_analyzed(self) -> None: + with patch("sw360.sw360_api.SW360Keycloak") as keycloak_class: + keycloak = keycloak_class.return_value + keycloak.get_keycloak_token.side_effect = [self._make_token(600), "not-a-jwt"] + + lib = SW360KeycloakClient(self.MYURL, "client-id", "client-secret") + lib._keycloak_token_provider._token = "not-a-jwt" + with self.assertRaises(SW360Error) as context: + lib.current_token + + self.assertTrue(context.exception.message.startswith("Unable to analyze token:")) + + def test_keycloak_client_uses_own_session(self) -> None: + token = self._make_token(600) + + with patch("sw360.sw360_api.SW360Keycloak") as keycloak_class: + keycloak_class.return_value.get_keycloak_token.return_value = token + lib = SW360KeycloakClient(self.MYURL, "client-id", "client-secret") + + self.assertIsNot(lib.session, session_default) + assert lib.session is not None + prepared = lib.session.prepare_request(requests.Request("GET", self.MYURL)) + self.assertEqual("Bearer " + token, prepared.headers["Authorization"]) + + def test_keycloak_client_current_token(self) -> None: + initial_token = self._make_token(600) + expiring_token = self._make_token(30) + fresh_token = self._make_token(600) + + with patch("sw360.sw360_api.SW360Keycloak") as keycloak_class: + keycloak = keycloak_class.return_value + keycloak.get_keycloak_token.side_effect = [initial_token, fresh_token] + + lib = SW360KeycloakClient(self.MYURL, "client-id", "client-secret") + self.assertEqual(initial_token, lib.current_token) + + lib._keycloak_token_provider._token = expiring_token + self.assertEqual(fresh_token, lib.current_token) + + @responses.activate + def test_keycloak_client_sends_renewed_token(self) -> None: + initial_token = self._make_token(600) + expiring_token = self._make_token(30) + fresh_token = self._make_token(600) + url = self.MYURL + "resource/api/releases/123" + + responses.add(responses.GET, url, body="{}", status=200, + content_type="application/json") + + with patch("sw360.sw360_api.SW360Keycloak") as keycloak_class: + keycloak = keycloak_class.return_value + keycloak.get_keycloak_token.side_effect = [initial_token, fresh_token] + + lib = SW360KeycloakClient(self.MYURL, "client-id", "client-secret") + lib._keycloak_token_provider._token = expiring_token + lib.api_get(url) + + self.assertEqual("Bearer " + fresh_token, + responses.calls[0].request.headers["Authorization"]) + + @responses.activate + def test_keycloak_client_sends_renewed_token_without_session(self) -> None: + initial_token = self._make_token(600) + expiring_token = self._make_token(30) + fresh_token = self._make_token(600) + url = self.MYURL + "resource/api/releases/123" + + responses.add(responses.GET, url, body="{}", status=200, + content_type="application/json") + + with patch("sw360.sw360_api.SW360Keycloak") as keycloak_class: + keycloak = keycloak_class.return_value + keycloak.get_keycloak_token.side_effect = [initial_token, fresh_token] + + lib = SW360KeycloakClient(self.MYURL, "client-id", "client-secret") + lib.force_no_session = True + lib._keycloak_token_provider._token = expiring_token + lib.api_get(url) + + self.assertEqual("Bearer " + fresh_token, + responses.calls[0].request.headers["Authorization"]) + def test_BaseMixin_constructor(self) -> None: lib = BaseMixin(self.MYURL, self.MYTOKEN, False) self.assertEqual(self.MYURL, lib.url)