Skip to content

Fix #241: Refresh Keycloak tokens during long operations #242

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
dhruvv16-hash wants to merge 3 commits into
base: main
Choose a base branch
from
Open

Fix #241: Refresh Keycloak tokens during long operations #242

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
4f98109
Refactor SW360 client login to central method
dhruvv16-hash Sep 15, 2026
08de381
Introduce KeycloakAuth to refresh SW360 tokens on 401
dhruvv16-hash Sep 15, 2026
00bda48
Refactor login logic to preserve original signature and introduce log…
dhruvv16-hash Sep 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 2 additions & 38 deletions capycli/bom/check_bom.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
from colorama import Fore, Style
from cyclonedx.model.bom import Bom
from cyclonedx.model.component import Component
from sw360 import SW360Error, SW360Keycloak
from sw360 import SW360Error

import capycli.common.script_base
from capycli.common.capycli_bom_support import CaPyCliBom, CycloneDxSupport
Expand Down Expand Up @@ -196,43 +196,7 @@ def run(self, args: Any) -> None:
if self._bom_has_items_without_id(bom):
print("There are SBOM items without Sw360 id - searching per name may take a little bit longer...")

if not args.sw360_token:
# command line argument precede environment variables
client_id = args.client_id
client_secret = args.client_secret

if not args.client_id and (not args.client_secret):
# look for environment variables
client_id = os.getenv("SW360Client_id")
client_secret = os.getenv("SW360Client_secret")
if client_id and client_secret and args.verbose:
print_text(" Found client id and client secret in environment variables.")

if client_id and client_secret:
url = args.sw360_url
if not url:
url = os.environ.get("SW360ServerUrl", "")
if not url:
print_red(" SW360 URL not specified!")
sys.exit(ResultCode.RESULT_COMMAND_ERROR)

if args.verbose:
print_text(" Creating token using client id and secret...")
kc = SW360Keycloak(url)
args.sw360_token = kc.get_keycloak_token(client_id, client_secret, write_access=False)
if args.sw360_token:
args.oauth2 = True
if args.verbose:
print_text(" Got token.")
else:
print_red(" Failed to get token!")
sys.exit(ResultCode.RESULT_AUTH_ERROR)

if args.sw360_token and args.oauth2 and args.verbose:
self.analyze_token(args.sw360_token)
print_text("")

if not self.login(token=args.sw360_token, url=args.sw360_url, oauth2=args.oauth2):
if not self.login_with_args(args, write_access=False):
print_red("ERROR: login failed!")
sys.exit(ResultCode.RESULT_AUTH_ERROR)

Expand Down
40 changes: 2 additions & 38 deletions capycli/bom/check_bom_item_status.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
from colorama import Fore, Style
from cyclonedx.model.bom import Bom
from cyclonedx.model.component import Component
from sw360 import SW360Error, SW360Keycloak
from sw360 import SW360Error

import capycli.common.script_base
from capycli.common.capycli_bom_support import CaPyCliBom, CycloneDxSupport
Expand Down Expand Up @@ -212,43 +212,7 @@ def run(self, args: Any) -> None:
if self._bom_has_items_without_id(bom):
print("There are SBOM items without Sw360 id - searching per name may take a little bit longer...")

if not args.sw360_token:
# command line argument precede environment variables
client_id = args.client_id
client_secret = args.client_secret

if not args.client_id and (not args.client_secret):
# look for environment variables
client_id = os.getenv("SW360Client_id")
client_secret = os.getenv("SW360Client_secret")
if client_id and client_secret and args.verbose:
print_text(" Found client id and client secret in environment variables.")

if client_id and client_secret:
url = args.sw360_url
if not url:
url = os.environ.get("SW360ServerUrl", "")
if not url:
print_red(" SW360 URL not specified!")
sys.exit(ResultCode.RESULT_COMMAND_ERROR)

if args.verbose:
print_text(" Creating token using client id and secret...")
kc = SW360Keycloak(url)
args.sw360_token = kc.get_keycloak_token(client_id, client_secret, write_access=False)
if args.sw360_token:
args.oauth2 = True
if args.verbose:
print_text(" Got token.")
else:
print_red(" Failed to get token!")
sys.exit(ResultCode.RESULT_AUTH_ERROR)

if args.sw360_token and args.oauth2 and args.verbose:
self.analyze_token(args.sw360_token)
print_text("")

if not self.login(token=args.sw360_token, url=args.sw360_url, oauth2=args.oauth2):
if not self.login_with_args(args, write_access=False):
print_red("ERROR: login failed!")
sys.exit(ResultCode.RESULT_AUTH_ERROR)

Expand Down
40 changes: 2 additions & 38 deletions capycli/bom/create_components.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
from cyclonedx.model.bom import Bom
from cyclonedx.model.component import Component
from cyclonedx.model.license import DisjunctiveLicense, LicenseExpression
from sw360 import SW360Error, SW360Keycloak
from sw360 import SW360Error

import capycli.common.json_support
import capycli.common.script_base
Expand Down Expand Up @@ -778,43 +778,7 @@ def run(self, args: Any) -> None:
print_text("Using relaxed debian version checks")
self.relaxed_debian_parsing = True

if not args.sw360_token:
# command line argument precede environment variables
client_id = args.client_id
client_secret = args.client_secret

if not args.client_id and (not args.client_secret):
# look for environment variables
client_id = os.getenv("SW360Client_id")
client_secret = os.getenv("SW360Client_secret")
if client_id and client_secret and args.verbose:
print_text(" Found client id and client secret in environment variables.")

if client_id and client_secret:
url = args.sw360_url
if not url:
url = os.environ.get("SW360ServerUrl", "")
if not url:
print_red(" SW360 URL not specified!")
sys.exit(ResultCode.RESULT_COMMAND_ERROR)

if args.verbose:
print_text(" Creating token using client id and secret...")
kc = SW360Keycloak(url)
args.sw360_token = kc.get_keycloak_token(client_id, client_secret, write_access=True)
if args.sw360_token:
args.oauth2 = True
if args.verbose:
print_text(" Got token.")
else:
print_red(" Failed to get token!")
sys.exit(ResultCode.RESULT_AUTH_ERROR)

if args.sw360_token and args.oauth2 and args.verbose:
self.analyze_token(args.sw360_token)
print_text("")

if not self.login(token=args.sw360_token, url=args.sw360_url, oauth2=args.oauth2):
if not self.login_with_args(args, write_access=True):
print_red("ERROR: login failed!")
sys.exit(ResultCode.RESULT_AUTH_ERROR)

Expand Down
42 changes: 3 additions & 39 deletions capycli/bom/findsources.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,6 @@
from cyclonedx.model.bom import Bom
from cyclonedx.model.component import Component
from sw360 import SW360Error
from sw360.sw360keycloak import SW360Keycloak

import capycli.common.script_base
from capycli import get_logger
Expand Down Expand Up @@ -772,44 +771,9 @@ def run(self, args: Any) -> None:
self.sw360_url = args.sw360_url

if self.sw360_url:
if not args.sw360_token:
# command line argument precede environment variables
client_id = args.client_id
client_secret = args.client_secret

if not args.client_id and (not args.client_secret):
# look for environment variables
client_id = os.getenv("SW360Client_id")
client_secret = os.getenv("SW360Client_secret")
if client_id and client_secret and args.verbose:
print_text(" Found client id and client secret in environment variables.")

if client_id and client_secret:
url = args.sw360_url
if not url:
url = os.environ.get("SW360ServerUrl", "")
if not url:
print_red(" SW360 URL not specified!")
sys.exit(ResultCode.RESULT_COMMAND_ERROR)

if args.verbose:
print_text(" Creating token using client id and secret...")
kc = SW360Keycloak(url)
args.sw360_token = kc.get_keycloak_token(client_id, client_secret, write_access=False)
if args.sw360_token:
args.oauth2 = True
if args.verbose:
print_text(" Got token.")
else:
print_red(" Failed to get token!")
sys.exit(ResultCode.RESULT_AUTH_ERROR)

if args.sw360_token and args.oauth2 and args.verbose:
self.analyze_token(args.sw360_token)
print_text("")

self.login(
token=args.sw360_token, url=self.sw360_url, oauth2=args.oauth2)
if not self.login_with_args(args, write_access=False):
print_red("ERROR: login failed!")
sys.exit(ResultCode.RESULT_AUTH_ERROR)
print("Using SW360 releases and components to detect GitHub url")
self.use_sw360 = True
else:
Expand Down
41 changes: 2 additions & 39 deletions capycli/bom/map_bom.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@
from cyclonedx.model.bom import Bom
from cyclonedx.model.component import Component
from packageurl import PackageURL
from sw360 import SW360, SW360Keycloak
from sw360 import SW360

import capycli.common.file_support
import capycli.common.script_base
Expand Down Expand Up @@ -935,44 +935,7 @@ def run(self, args: Any) -> None:
if self.verbosity > 1:
print_text(" ", self.get_comp_count_text(sbom), "read from SBOM")

if not args.sw360_token:
# command line argument precede environment variables
client_id = args.client_id
client_secret = args.client_secret

if not args.client_id and (not args.client_secret):
# look for environment variables
client_id = os.getenv("SW360Client_id")
client_secret = os.getenv("SW360Client_secret")
if client_id and client_secret and args.verbose:
print_text("\n Found client id and client secret in environment variables.")

if client_id and client_secret:
url = args.sw360_url
if not url:
url = os.environ.get("SW360ServerUrl", "")
if not url:
print_red(" SW360 URL not specified!")
sys.exit(ResultCode.RESULT_COMMAND_ERROR)

if args.verbose:
print_text(" Creating token using client id and secret...")
kc = SW360Keycloak(url)
args.sw360_token = kc.get_keycloak_token(client_id, client_secret, write_access=False)
if args.sw360_token:
args.oauth2 = True
if args.verbose:
print_text(" Got token.")
else:
print_red(" Failed to get token!")
sys.exit(ResultCode.RESULT_AUTH_ERROR)

if args.sw360_token and args.oauth2 and args.verbose:
self.analyze_token(args.sw360_token)
print_text("")

print_text(" Checking access to SW360...")
if not self.login(token=args.sw360_token, url=args.sw360_url, oauth2=args.oauth2):
if not self.login_with_args(args, write_access=False):
print_red("ERROR: login failed!")
sys.exit(ResultCode.RESULT_AUTH_ERROR)

Expand Down
89 changes: 89 additions & 0 deletions capycli/common/keycloak_auth.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,89 @@
import logging
from datetime import datetime
from typing import Any

import jwt
from sw360 import SW360Keycloak
from requests.auth import AuthBase
from requests.models import PreparedRequest, Response

from capycli import get_logger
from capycli.common.print import print_yellow

LOG = get_logger(__name__)


class KeycloakAuth(AuthBase):
def __init__(self, url: str, client_id: str, client_secret: str, write_access: bool, initial_token: str) -> None:
self.kc = SW360Keycloak(url)
self.client_id = client_id
self.client_secret = client_secret
self.write_access = write_access
self.token = initial_token

def _refresh_token(self) -> bool:
try:
new_token = self.kc.get_keycloak_token(self.client_id, self.client_secret, self.write_access)
if new_token:
self.token = new_token
if LOG.isEnabledFor(logging.DEBUG):
LOG.debug("SW360 Keycloak token refreshed successfully.")
return True
else:
print_yellow(" Failed to refresh token: empty token returned")
except Exception:
print_yellow(" Failed to refresh token. Check credentials and server status.")
return False

def is_token_expiring_soon(self) -> bool:
try:
decoded = jwt.decode( # type: ignore
self.token, algorithms=["HS256"], options={"verify_signature": False})
if "exp" in decoded:
exp = datetime.fromtimestamp(int(decoded["exp"]))
# refresh if less than 5 minutes remaining
return (exp - datetime.now()).total_seconds() < 300
except Exception:
pass
return False

def __call__(self, r: PreparedRequest) -> PreparedRequest:
# Refresh token shortly before expiry
if self.token and self.is_token_expiring_soon():
self._refresh_token()

if self.token:
r.headers['Authorization'] = 'Bearer ' + self.token

r.register_hook('response', self.handle_401) # type: ignore
return r

def handle_401(self, r: Response, **kwargs: Any) -> Response:
if r.status_code == 401 and not getattr(r.request, '_sw360_retried', False):
# Only retry safe, idempotent methods to prevent duplicating operations
if r.request and r.request.method not in ["GET", "HEAD", "OPTIONS"]:
return r

if LOG.isEnabledFor(logging.DEBUG):
LOG.debug("Received 401 Unauthorized, attempting token refresh...")

if self._refresh_token():
# Consume content of response so we can reuse the connection
_ = r.content
r.close()

# Create a new request based on the old one
assert r.request is not None

new_req = r.request.copy()
new_req.headers['Authorization'] = 'Bearer ' + self.token
setattr(new_req, '_sw360_retried', True)

# Send new request and get new response
if hasattr(r, "connection") and r.connection:
new_resp = r.connection.send(new_req, **kwargs)
new_resp.history.append(r)
new_resp.request = new_req
return new_resp

return r
Loading
Loading