diff --git a/capycli/bom/check_bom.py b/capycli/bom/check_bom.py index 3195b5ca..3c2850a3 100644 --- a/capycli/bom/check_bom.py +++ b/capycli/bom/check_bom.py @@ -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 @@ -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) diff --git a/capycli/bom/check_bom_item_status.py b/capycli/bom/check_bom_item_status.py index 1374fbf5..db724d3d 100644 --- a/capycli/bom/check_bom_item_status.py +++ b/capycli/bom/check_bom_item_status.py @@ -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 @@ -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) diff --git a/capycli/bom/create_components.py b/capycli/bom/create_components.py index 60a9a4a1..3bd81389 100644 --- a/capycli/bom/create_components.py +++ b/capycli/bom/create_components.py @@ -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 @@ -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) diff --git a/capycli/bom/findsources.py b/capycli/bom/findsources.py index 0d32df02..178ab3d7 100644 --- a/capycli/bom/findsources.py +++ b/capycli/bom/findsources.py @@ -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 @@ -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: diff --git a/capycli/bom/map_bom.py b/capycli/bom/map_bom.py index 357ffaab..5056b37f 100644 --- a/capycli/bom/map_bom.py +++ b/capycli/bom/map_bom.py @@ -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 @@ -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) diff --git a/capycli/common/keycloak_auth.py b/capycli/common/keycloak_auth.py new file mode 100644 index 00000000..3f002d5a --- /dev/null +++ b/capycli/common/keycloak_auth.py @@ -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 diff --git a/capycli/common/script_base.py b/capycli/common/script_base.py index c256ce7f..1d9c5d00 100644 --- a/capycli/common/script_base.py +++ b/capycli/common/script_base.py @@ -65,11 +65,62 @@ def login(self, token: str = "", url: str = "", oauth2: bool = False) -> bool: print_red(" You are not authorized!") sys.exit(ResultCode.RESULT_AUTH_ERROR) else: - print_red(" Error authorizing user: " + repr(swex)) - sys.exit(ResultCode.RESULT_AUTH_ERROR) + print_red(" Error accessing SW360 API!") + sys.exit(ResultCode.RESULT_ERROR_ACCESSING_SW360) return result + def login_with_args(self, app_args: Any, write_access: bool = False) -> bool: + """Login to SW360 handling Keycloak OAuth2 with app_args""" + url = getattr(app_args, "sw360_url", "") + if not url: + url = os.environ.get("SW360ServerUrl", "") + + token = getattr(app_args, "sw360_token", "") + oauth2 = getattr(app_args, "oauth2", False) + keycloak_auth = None + + if not token: + client_id = getattr(app_args, "client_id", "") + client_secret = getattr(app_args, "client_secret", "") + if not client_id and not client_secret: + client_id = os.getenv("SW360Client_id") + client_secret = os.getenv("SW360Client_secret") + if client_id and client_secret and getattr(app_args, "verbose", False): + print_text(" Found client id and client secret in environment variables.") + + if client_id and client_secret: + if not url: + print_red(" SW360 URL not specified!") + sys.exit(ResultCode.RESULT_COMMAND_ERROR) + if getattr(app_args, "verbose", False): + print_text(" Creating token using client id and secret...") + + from sw360 import SW360Keycloak + kc = SW360Keycloak(url) + kc_token = kc.get_keycloak_token(client_id, client_secret, write_access=write_access) + if kc_token: + token = kc_token + oauth2 = True + if hasattr(app_args, "sw360_token"): + app_args.sw360_token = kc_token + if hasattr(app_args, "oauth2"): + app_args.oauth2 = True + if getattr(app_args, "verbose", False): + print_text(" Got token.") + + # Introduce abstraction for refreshing Keycloak credentials + from capycli.common.keycloak_auth import KeycloakAuth + keycloak_auth = KeycloakAuth(url, client_id, client_secret, write_access, kc_token) + else: + print_red(" Failed to get token!") + sys.exit(ResultCode.RESULT_AUTH_ERROR) + + success = self.login(token, url, oauth2) + if success and keycloak_auth and self.client: + self.client.keycloak_auth = keycloak_auth + return success + def analyze_token(self, token: str) -> None: """Analyzes the user provided token. If we can decode it, then it is an OAuth2 token.""" diff --git a/capycli/project/check_prerequisites.py b/capycli/project/check_prerequisites.py index ac1e9db5..50db57d9 100644 --- a/capycli/project/check_prerequisites.py +++ b/capycli/project/check_prerequisites.py @@ -13,7 +13,6 @@ from colorama import Fore from cyclonedx.model.bom import Bom -from sw360 import SW360Keycloak import capycli.common.script_base from capycli import get_logger @@ -315,43 +314,7 @@ def run(self, args: Any) -> None: print(" -client_secret CLIENT_SECRET the SW360 client_secret to be used for token generation") return - 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) diff --git a/capycli/project/create_bom.py b/capycli/project/create_bom.py index dd7bb0db..5778d6a2 100644 --- a/capycli/project/create_bom.py +++ b/capycli/project/create_bom.py @@ -1,4 +1,4 @@ -# ------------------------------------------------------------------------------- +# ------------------------------------------------------------------------------- # Copyright (c) 2021-2026 Siemens # All Rights Reserved. # Author: thomas.graf@siemens.com @@ -14,7 +14,7 @@ from cyclonedx.model.bom import Bom from cyclonedx.model.component import Component from packageurl import PackageURL -from sw360 import SW360Error, SW360Keycloak +from sw360 import SW360Error import capycli.common.script_base from capycli import get_logger @@ -194,21 +194,7 @@ def run(self, args: Any) -> None: self.show_command_help() return - if not args.sw360_token and args.client_id and args.client_secret: - print_text("Creating token using client id and secret...") - kc = SW360Keycloak(args.sw360_url) - args.sw360_token = kc.get_keycloak_token(args.client_id, args.client_secret, write_access=False) - if args.sw360_token: - args.oauth2 = True - print_text(" Got token.") - else: - print_red(" Failed to get token!") - sys.exit(ResultCode.RESULT_AUTH_ERROR) - - if args.sw360_token and args.oauth2: - self.analyze_token(args.sw360_token) - - 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) diff --git a/capycli/project/create_project.py b/capycli/project/create_project.py index cff1b7f2..276b63a8 100644 --- a/capycli/project/create_project.py +++ b/capycli/project/create_project.py @@ -13,7 +13,7 @@ import requests from cyclonedx.model.bom import Bom -from sw360 import SW360Error, SW360Keycloak +from sw360 import SW360Error import capycli.common.script_base from capycli import get_logger @@ -374,43 +374,7 @@ def run(self, args: Any) -> None: print_text("Project version will be updated with version: " + args.old_version) is_update_version = 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) diff --git a/capycli/project/find_project.py b/capycli/project/find_project.py index ee958594..66ebf3a5 100644 --- a/capycli/project/find_project.py +++ b/capycli/project/find_project.py @@ -7,13 +7,12 @@ # ------------------------------------------------------------------------------- import logging -import os import sys import traceback from typing import Any, Dict, Optional import requests -from sw360 import SW360Error, SW360Keycloak +from sw360 import SW360Error import capycli.common.script_base from capycli.common.print import print_red, print_text, print_yellow @@ -123,43 +122,7 @@ def run(self, args: Any) -> None: print(" -client_secret CLIENT_SECRET the SW360 client_secret to be used for token generation") return - 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) diff --git a/capycli/project/get_license_info.py b/capycli/project/get_license_info.py index 6e2684bd..51ed8262 100644 --- a/capycli/project/get_license_info.py +++ b/capycli/project/get_license_info.py @@ -12,7 +12,7 @@ import sys from typing import Any, Dict, List -from sw360 import SW360Error, SW360Keycloak +from sw360 import SW360Error import capycli.common.script_base from capycli.common.json_support import load_json_file @@ -241,43 +241,7 @@ def run(self, args: Any) -> None: print_red("Input file not found!") sys.exit(ResultCode.RESULT_FILE_NOT_FOUND) - 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) diff --git a/capycli/project/project_component_check.py b/capycli/project/project_component_check.py index 5f628515..b8a47bb7 100644 --- a/capycli/project/project_component_check.py +++ b/capycli/project/project_component_check.py @@ -7,11 +7,10 @@ # ------------------------------------------------------------------------------- import logging -import os import sys from typing import Any -from sw360 import SW360Error, SW360Keycloak +from sw360 import SW360Error import capycli.common.script_base from capycli.bom.component_check import ComponentCheck @@ -151,43 +150,7 @@ def run(self, args: Any) -> None: self.component_check.files_to_ignore = self.component_check.component_check_list.get("files_to_ignore", []) print_text(f" {len(self.component_check.files_to_ignore)} components will be ignored.") - 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) diff --git a/capycli/project/show_ecc.py b/capycli/project/show_ecc.py index 01bca91b..90af3f83 100644 --- a/capycli/project/show_ecc.py +++ b/capycli/project/show_ecc.py @@ -7,11 +7,10 @@ # ------------------------------------------------------------------------------- import logging -import os import sys from typing import Any, Dict -from sw360 import SW360Error, SW360Keycloak +from sw360 import SW360Error import capycli.common.script_base from capycli.common.json_support import write_json_to_file @@ -182,43 +181,7 @@ def run(self, args: Any) -> None: print(" -v be verbose") return - 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) diff --git a/capycli/project/show_licenses.py b/capycli/project/show_licenses.py index 44ed6272..fcb28b5e 100644 --- a/capycli/project/show_licenses.py +++ b/capycli/project/show_licenses.py @@ -15,7 +15,6 @@ from cli_support import CliFile from colorama import Fore, Style -from sw360 import SW360Keycloak import capycli.common.script_base from capycli.common.print import print_red, print_text, print_yellow @@ -209,43 +208,7 @@ def run(self, args: Any) -> None: self.show_command_help() return - 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) diff --git a/capycli/project/show_project.py b/capycli/project/show_project.py index 715f6541..192ea00c 100644 --- a/capycli/project/show_project.py +++ b/capycli/project/show_project.py @@ -7,12 +7,11 @@ # ------------------------------------------------------------------------------- import logging -import os import sys from typing import Any, Dict, Optional from colorama import Fore -from sw360 import SW360Error, SW360Keycloak +from sw360 import SW360Error import capycli.common.json_support import capycli.common.script_base @@ -202,43 +201,7 @@ def run(self, args: Any) -> None: print(" -v be verbose") return - 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) diff --git a/capycli/project/show_vulnerabilities.py b/capycli/project/show_vulnerabilities.py index accdaacf..81f2e97f 100644 --- a/capycli/project/show_vulnerabilities.py +++ b/capycli/project/show_vulnerabilities.py @@ -7,13 +7,12 @@ # ------------------------------------------------------------------------------- import logging -import os import sys from typing import Any, Dict, Optional import requests from colorama import Fore, Style -from sw360 import SW360Error, SW360Keycloak +from sw360 import SW360Error import capycli.common.json_support import capycli.common.script_base @@ -248,43 +247,7 @@ def run(self, args: Any) -> None: if args.verbose: print("Output format is", self.format) - 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) diff --git a/tests/test_keycloak_auth.py b/tests/test_keycloak_auth.py new file mode 100644 index 00000000..f4ac7467 --- /dev/null +++ b/tests/test_keycloak_auth.py @@ -0,0 +1,94 @@ +from unittest.mock import MagicMock, patch + +from requests.models import PreparedRequest, Response + +from capycli.common.keycloak_auth import KeycloakAuth + + +class TestKeycloakAuth: + def test_refresh_token(self) -> None: + with patch('capycli.common.keycloak_auth.SW360Keycloak') as mock_kc: + instance = mock_kc.return_value + instance.get_keycloak_token.return_value = "new_token" + + auth = KeycloakAuth("http://localhost", "client", "secret", False, "old_token") + assert auth._refresh_token() is True + assert auth.token == "new_token" + + def test_call_adds_header(self) -> None: + with patch('capycli.common.keycloak_auth.SW360Keycloak') as _: + auth = KeycloakAuth("http://localhost", "client", "secret", False, "old_token") + # Instead of assigning to a method, patch it: + with patch.object(auth, 'is_token_expiring_soon', return_value=False): + req = PreparedRequest() + setattr(req, 'headers', {}) + setattr(req, 'register_hook', MagicMock()) + + req2 = auth(req) + assert getattr(req2, 'headers')['Authorization'] == 'Bearer old_token' + + def test_handle_401_get(self) -> None: + with patch('capycli.common.keycloak_auth.SW360Keycloak') as mock_kc: + instance = mock_kc.return_value + instance.get_keycloak_token.return_value = "new_token" + + auth = KeycloakAuth("http://localhost", "client", "secret", False, "old_token") + + r = Response() + r.status_code = 401 + r.request = PreparedRequest() + r.request.method = "GET" + r.request.url = "http://localhost" + setattr(r.request, 'headers', {}) + setattr(r, 'connection', MagicMock()) + + mock_new_resp = Response() + mock_new_resp.status_code = 200 + getattr(r, 'connection').send.return_value = mock_new_resp + + new_r = auth.handle_401(r) + + assert getattr(r, 'connection').send.called + assert new_r == mock_new_resp + + def test_handle_401_no_retry_for_post(self) -> None: + with patch('capycli.common.keycloak_auth.SW360Keycloak') as mock_kc: + instance = mock_kc.return_value + instance.get_keycloak_token.return_value = "new_token" + + auth = KeycloakAuth("http://localhost", "client", "secret", False, "old_token") + + r = Response() + r.status_code = 401 + r.request = PreparedRequest() + r.request.method = "POST" + r.request.url = "http://localhost" + setattr(r.request, 'headers', {}) + setattr(r, 'connection', MagicMock()) + + new_r = auth.handle_401(r) + + assert not getattr(r, 'connection').send.called + assert new_r == r + + def test_handle_401_no_second_retry(self) -> None: + with patch('capycli.common.keycloak_auth.SW360Keycloak') as mock_kc: + instance = mock_kc.return_value + instance.get_keycloak_token.return_value = "new_token" + + auth = KeycloakAuth("http://localhost", "client", "secret", False, "old_token") + + r = Response() + r.status_code = 401 + r.request = PreparedRequest() + r.request.method = "GET" + r.request.url = "http://localhost" + setattr(r.request, 'headers', {}) + # This simulates a request that was already retried + setattr(r.request, '_sw360_retried', True) + setattr(r, 'connection', MagicMock()) + + new_r = auth.handle_401(r) + + assert not getattr(r, 'connection').send.called + assert new_r == r