diff --git a/pyproject.toml b/pyproject.toml index 92b2d82..c969f5a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -37,6 +37,7 @@ dependencies = [ "XlsxWriter", "PyYAML", "fosslight_util>=2.2.8", + "truststore", "python-magic-bin; sys_platform == 'win32'", "python-magic; 'darwin' in sys_platform", "python-magic; 'linux' in sys_platform", diff --git a/src/fosslight_binary/_binary_dao.py b/src/fosslight_binary/_binary_dao.py index 4b34a4c..8c04531 100755 --- a/src/fosslight_binary/_binary_dao.py +++ b/src/fosslight_binary/_binary_dao.py @@ -7,6 +7,7 @@ import json import logging import os +import ssl import urllib.error import urllib.request from typing import Dict, List, Optional, Tuple @@ -43,6 +44,31 @@ def _get_chunk_size() -> int: _CHUNK_SIZE = _get_chunk_size() MatchKey = Tuple[str, str] +_KB_SSL_VERIFY_FALSE = {"0", "false", "no", "off"} +_logged_insecure_kb_ssl = False + + +def kb_ssl_verify_enabled() -> bool: + raw = os.environ.get("KB_SSL_VERIFY", "true") + return raw.strip().lower() not in _KB_SSL_VERIFY_FALSE + + +def create_kb_ssl_context() -> ssl.SSLContext: + """TLS context for KB HTTPS. Uses the OS trust store (Windows store, like the browser).""" + global _logged_insecure_kb_ssl + if not kb_ssl_verify_enabled(): + if not _logged_insecure_kb_ssl: + logger.warning("KB TLS certificate verification is disabled (KB_SSL_VERIFY)") + _logged_insecure_kb_ssl = True + ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + return ctx + try: + import truststore + return truststore.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + except ImportError: + return ssl.create_default_context() def resolve_kb_config(kb_url: str = "", kb_token: str = "") -> Tuple[str, str]: @@ -78,7 +104,9 @@ def check_binary_match_endpoint(kb_url: str, kb_token: str = "") -> Tuple[bool, request.add_header("Authorization", f"Bearer {kb_token}") try: - with urllib.request.urlopen(request, timeout=_PROBE_TIMEOUT_SEC) as response: + with urllib.request.urlopen( + request, timeout=_PROBE_TIMEOUT_SEC, context=create_kb_ssl_context() + ) as response: response.read() return True, "" except urllib.error.HTTPError as ex: @@ -257,7 +285,9 @@ def _post_binary_match(kb_url: str, kb_token: str, items: list) -> Optional[dict request.add_header("Authorization", f"Bearer {kb_token}") try: - with urllib.request.urlopen(request, timeout=_HTTP_TIMEOUT_SEC) as response: + with urllib.request.urlopen( + request, timeout=_HTTP_TIMEOUT_SEC, context=create_kb_ssl_context() + ) as response: body = response.read().decode() return json.loads(body) if body else {} except urllib.error.HTTPError as ex: diff --git a/src/fosslight_binary/_help.py b/src/fosslight_binary/_help.py index 7045551..4648f81 100644 --- a/src/fosslight_binary/_help.py +++ b/src/fosslight_binary/_help.py @@ -34,6 +34,8 @@ ──────────────────────────────────────────────────────────────────── --kb_url KB API URL (priority: parameter > KB_URL env > default) --kb_token KB bearer token (priority: parameter > KB_TOKEN env) + HTTPS uses the OS certificate store. Set KB_SSL_VERIFY=false + only if certificate verification must be skipped. --notice Print the open source license notice text --no_correction Skip OSS information correction with sbom-info.yaml --correct_fpath Path to custom sbom-info.yaml file diff --git a/src/fosslight_binary/_jar_analysis.py b/src/fosslight_binary/_jar_analysis.py index 4f41fbb..8a0bcc4 100644 --- a/src/fosslight_binary/_jar_analysis.py +++ b/src/fosslight_binary/_jar_analysis.py @@ -20,8 +20,8 @@ logger = logging.getLogger(constant.LOGGER_NAME) _CENTRAL_SEARCH_URL = "https://search.maven.org/solrsearch/select" -_REQUEST_TIMEOUT = 10 # seconds – used for HEAD / POM download -_CENTRAL_SEARCH_TIMEOUT = 2.5 # seconds – tight timeout for Search API (retried on timeout) +_REQUEST_TIMEOUT = 10 # seconds - used for HEAD / POM download +_CENTRAL_SEARCH_TIMEOUT = 2.5 # seconds - tight timeout for Search API (retried on timeout) _MAVEN_JAR_HTTP_TIMEOUT = (2, 2) # match Util probe timeouts for multi-repo jar checks _MAX_RETRY = 3 # maximum Central API retry attempts per JAR _central_network_warned = False # Flag to suppress repeated network-unavailable warnings within one run @@ -141,7 +141,7 @@ def _search_central_by_sha1(sha1, timeout=None): "version": version, }, False except requests.exceptions.Timeout: - logger.debug(f"Maven Central SHA-1 search timed out ({sha1}) – will retry") + logger.debug(f"Maven Central SHA-1 search timed out ({sha1}) - will retry") return {}, True except Exception as ex: if _is_network_error(ex): @@ -169,7 +169,7 @@ def _download_pom_to_tempfile(group_id, artifact_id, version, timeout=None): logger.debug(f"POM downloaded to {tmp.name} from {url}") return tmp.name, False except requests.exceptions.Timeout: - logger.debug(f"POM download timed out from {url} – will retry") + logger.debug(f"POM download timed out from {url} - will retry") any_timeout = True except Exception as ex: if _is_network_error(ex): @@ -241,7 +241,7 @@ def _process_one_jar(jar_path, rel_path, sha1, search_timeout=None, skip_central else: central_info, timed_out = _search_central_by_sha1(sha1, timeout=search_timeout) if timed_out: - logger.debug(f"{rel_path}: Central SHA-1 search timed out – will retry") + logger.debug(f"{rel_path}: Central SHA-1 search timed out - will retry") return None, True g2, a2, v2, url2, pom_tmp_path = _read_pom_from_jar(jar_path) @@ -256,7 +256,7 @@ def _process_one_jar(jar_path, rel_path, sha1, search_timeout=None, skip_central names_match = (central_oss_name == jar_oss_name and c_version == v2) if names_match: - logger.debug(f"{rel_path}: Central and JAR pom.xml match ({central_oss_name} {c_version}) – using JAR pom.xml for license") + logger.debug(f"{rel_path}: Central and JAR pom.xml match ({central_oss_name} {c_version}) - using JAR pom.xml for license") groupId, artifactId, version, project_url = g2, a2, v2, url2 source = 'pom.xml' confirmed_in_central = True @@ -286,7 +286,7 @@ def _process_one_jar(jar_path, rel_path, sha1, search_timeout=None, skip_central tmp_path, timed_out = _download_pom_to_tempfile( groupId, artifactId, version, timeout=search_timeout) if timed_out: - logger.debug(f"{rel_path}: POM download timed out – will retry") + logger.debug(f"{rel_path}: POM download timed out - will retry") if pom_tmp_path: try: os.remove(pom_tmp_path) @@ -314,7 +314,7 @@ def _process_one_jar(jar_path, rel_path, sha1, search_timeout=None, skip_central pass else: - logger.debug(f"{rel_path}: not found in Maven Central – falling back to JAR internals") + logger.debug(f"{rel_path}: not found in Maven Central - falling back to JAR internals") if g2 or a2: groupId, artifactId, version, project_url = g2, a2, v2, url2 @@ -393,7 +393,7 @@ def analyze_jar_file(path_to_find_bin, path_to_exclude): jar_files.append(os.path.join(root_dir, fname)) if not jar_files: - logger.info("No .jar files found – skipping JAR OSS analysis.") + logger.info("No .jar files found - skipping JAR OSS analysis.") return jar_items, success for jar_path in jar_files: @@ -409,7 +409,7 @@ def analyze_jar_file(path_to_find_bin, path_to_exclude): jar_path, rel_path, sha1, search_timeout=_CENTRAL_SEARCH_TIMEOUT) if needs_retry: - logger.debug(f"{rel_path}: Central API timed out – queued for retry (attempt 1/{_MAX_RETRY})") + logger.debug(f"{rel_path}: Central API timed out - queued for retry (attempt 1/{_MAX_RETRY})") pending_sha1s.add(sha1) retry_queue.append((jar_path, rel_path, sha1, 1)) elif result is not None: @@ -421,7 +421,7 @@ def analyze_jar_file(path_to_find_bin, path_to_exclude): if attempt >= _MAX_RETRY: logger.warning( f"{rel_path}: Maven Central API timed out after {_MAX_RETRY} attempts" - " – falling back to JAR internals") + " - falling back to JAR internals") result, _ = _process_one_jar(jar_path, rel_path, sha1, skip_central_search=True) if result is not None: _store_jar_result(jar_items, sha1, result) diff --git a/tests/test_kb_ssl.py b/tests/test_kb_ssl.py new file mode 100644 index 0000000..a52ad19 --- /dev/null +++ b/tests/test_kb_ssl.py @@ -0,0 +1,54 @@ +# Copyright (c) 2026 LG Electronics Inc. +# SPDX-License-Identifier: Apache-2.0 +"""Tests for KB HTTPS SSL context selection.""" + +import ssl +from unittest.mock import patch + +from fosslight_binary._binary_dao import create_kb_ssl_context, kb_ssl_verify_enabled + + +def test_kb_ssl_verify_enabled_default(monkeypatch): + monkeypatch.delenv("KB_SSL_VERIFY", raising=False) + assert kb_ssl_verify_enabled() is True + + +def test_kb_ssl_verify_disabled_values(monkeypatch): + for value in ("false", "0", "no", "OFF"): + monkeypatch.setenv("KB_SSL_VERIFY", value) + assert kb_ssl_verify_enabled() is False + + +def test_create_kb_ssl_context_insecure(monkeypatch): + monkeypatch.setenv("KB_SSL_VERIFY", "false") + ctx = create_kb_ssl_context() + assert ctx.verify_mode == ssl.CERT_NONE + assert ctx.check_hostname is False + + +def test_create_kb_ssl_context_uses_truststore_when_verify_on(monkeypatch): + monkeypatch.setenv("KB_SSL_VERIFY", "true") + ctx = create_kb_ssl_context() + assert ctx.verify_mode != ssl.CERT_NONE + assert ctx.check_hostname is True + + +def test_probe_passes_ssl_context(): + from fosslight_binary._binary_dao import check_binary_match_endpoint + + class _Resp: + def read(self): + return b"{}" + + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + with patch("fosslight_binary._binary_dao.urllib.request.urlopen", return_value=_Resp()) as urlopen: + available, comment = check_binary_match_endpoint("https://kb.example/", "") + assert available is True + assert comment == "" + assert urlopen.call_args.kwargs["context"] is not None + assert isinstance(urlopen.call_args.kwargs["context"], ssl.SSLContext)