Use os trust store for https kb

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",

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:

src/fosslight_binary/_help.py

@@ -34,6 +34,8 @@
     ────────────────────────────────────────────────────────────────────
     --kb_url <url>         KB API URL (priority: parameter > KB_URL env > default)
     --kb_token <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> Path to custom sbom-info.yaml file

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)

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)