Replace Binary DB direct access with ldb_service API

pyproject.toml

@@ -32,7 +32,6 @@ dependencies = [
     "numpy",
     "pandas",
     "parmap",
-    "psycopg2-binary>=2.9.10",
     "python-dateutil",
     "py-tlsh",
     "pytz",

src/fosslight_android/_binary_db_controller.py

@@ -2,143 +2,177 @@
 # -*- coding: utf-8 -*-
 # SPDX-FileCopyrightText: Copyright 2023 LG Electronics Inc.
 # SPDX-License-Identifier: Apache-2.0
+"""Binary DB lookup via ldb_service POST /binary/match."""
+import json
 import logging
 import os
-import psycopg2
-import pandas as pd
-import tlsh
+import urllib.error
+import urllib.request
+from typing import Dict, List, Optional, Tuple
+
 from ._common import CONST_TLSH_NULL
 from fosslight_util.constant import LOGGER_NAME
 
 logger = logging.getLogger(LOGGER_NAME)
-columns = ['filename', 'pathname', 'checksum', 'tlshchecksum', 'ossname', 'ossversion', 'license', 'platformname',
-           'platformversion']
-
-DB_USER = 'bin_analysis_script_user'
-DB_PSWD = 'script_123'
-
-
-def connect_to_lge_bin_db():
-    conn = ""
-    cur = ""
-    user = DB_USER
-    password = DB_PSWD
-    host_product = 'bat.lge.com'
-    dbname = 'bat'
-    port = '5432'
-
-    connection_string = "dbname={dbname} user={user} host={host} password={password} port={port}" \
-        .format(dbname=dbname,
-                user=user,
-                host=host_product,
-                password=password,
-                port=port)
-    try:
-        conn = psycopg2.connect(connection_string)
-        cur = conn.cursor()
-    except Exception:
-        conn = ""
-        cur = ""
-    return conn, cur
-
-
-def get_oss_info_from_db(platform_version, bin_info_list, return_list):
-
-    conn, cur = connect_to_lge_bin_db()
-    if conn != "" and cur != "":
-        for item in bin_info_list:
-            try:
-                checksum_value = item.checksum
-                tlsh_value = item.tlsh
-                bin_file = item.bin_name
-                bin_file_name = os.path.basename(bin_file)
-                # Get OSS Information From Binary DB by matching checksum , filename, tlsh
-                df_result, item_comment, is_new = get_oss_info_by_tlsh_and_filename(bin_file_name,
-                                                                                    checksum_value, tlsh_value,
-                                                                                    item.source_code_path,
-                                                                                    platform_version, conn, cur)
-                item.set_comment(item_comment)
-                item.is_new_bin = is_new
-                if df_result is not None and len(df_result) > 0:
-                    for idx, row in df_result.iterrows():
-                        if idx == 0:
-                            item.set_oss_name(row['ossname'])
-                            item.set_oss_version(row['ossversion'])
-                            item.set_license(row['license'])
-                        else:  # In case more than 2 OSS is used for this bin.
-                            item.set_additional_oss_items(row['ossname'] + '\t' + row['ossversion'] + '\t' + row['license'])
-            except Exception as error:
-                logger.warn(f"READ OSS :{error}")
-
-        disconnect_lge_bin_db(conn, cur)
-    return_list.extend(bin_info_list)
-
-
-def get_oss_info_by_tlsh_and_filename(file_name, checksum_value, tlsh_value, source_path, platform_version, conn, cur):
-    sql_statement = "SELECT filename,pathname,checksum,tlshchecksum,ossname,ossversion,license,platformname,platformversion FROM lgematching "
-    sql_statement_checksum = " WHERE filename=%(fname)s AND checksum=%(checksum)s;"
-    sql_checksum_params = {'fname': file_name, 'checksum': checksum_value}
-    sql_statement_filename = "SELECT tlshchecksum FROM lgematching WHERE filename=%(fname)s AND tlshchecksum <> '0' ORDER BY ( " \
-                             "CASE " \
-                             "WHEN sourcepath = %(src_path)s AND lower(platformname)=%(plat_name)s " \
-                             "AND platformversion=%(plat_version)s THEN 1 " \
-                             "WHEN sourcepath = %(src_path)s AND lower(platformname)=%(plat_name)s THEN 2 " \
-                             "WHEN lower(platformname)=%(plat_name)s AND platformversion=%(plat_version)s THEN 3 " \
-                             "WHEN lower(platformname)=%(plat_name)s THEN 4 " \
-                             "ELSE 5 " \
-                             "END), updatedate DESC;"
-    sql_filename_params = {'fname': file_name, 'src_path': source_path, 'plat_version': platform_version, 'plat_name': "android"}
-    auto_id_comment = ""
-    final_result_item = ""
-    is_new = False
-
-    # Match checksum and fileName
-    df_result = get_list_by_using_query(sql_statement + sql_statement_checksum, sql_checksum_params, columns, conn, cur)
-    if df_result is not None and len(df_result) > 0:  # Found a file with the same checksum.
-        final_result_item = df_result
-    else:  # Can't find files that have same name and checksum
-        # Match tlsh and fileName
-        df_result = get_list_by_using_query(sql_statement_filename, sql_filename_params, ['tlshchecksum'], conn, cur)
-        if df_result is None or len(df_result) <= 0:
-            final_result_item = ""
-            auto_id_comment = "New Binary/"
-            is_new = True
-        elif tlsh_value == CONST_TLSH_NULL:  # Couldn't get the tlsh of a file.
-            final_result_item = ""
-        else:
-            matched_tlsh = ""
-            for row in df_result.tlshchecksum:
-                try:
-                    if row != CONST_TLSH_NULL:
-                        tlsh_diff = tlsh.diff(row, tlsh_value)
-                        if tlsh_diff <= 120:  # MATCHED
-                            matched_tlsh = row
-                            break
-                except Exception as error:  # TLSH COMPARISON FAILED
-                    logger.debug(f"Comparing TLSH:{error}")
 
-            if matched_tlsh != "":
-                final_result_item = get_list_by_using_query(
-                    sql_statement + " WHERE filename=%(fname)s AND tlshchecksum=%(tlsh)s;", {'fname': file_name, 'tlsh': matched_tlsh},
-                    columns, conn, cur)
+DEFAULT_KB_URL = "http://fosslight-kb.lge.com/"
+_BINARY_MATCH_PATH = "/binary/match"
+_HTTP_TIMEOUT_SEC = 120
+_CHUNK_SIZE = int(os.environ.get("BINARY_MATCH_CHUNK_SIZE", "1000"))
+
+MatchKey = Tuple[str, str]
+# (response_or_None, unreachable) — unreachable stops remaining chunks
+PostMatchResult = Tuple[Optional[dict], bool]
+
+
+def resolve_kb_config(kb_url: str = "", kb_token: str = "") -> Tuple[str, str]:
+    url = (kb_url or os.environ.get("KB_URL", DEFAULT_KB_URL)).strip() or DEFAULT_KB_URL
+    token = (kb_token or "").strip() or (os.environ.get("KB_TOKEN") or "").strip()
+    return f"{url.rstrip('/')}/", token
+
+
+def _item_filename(item) -> str:
+    return item.binary_name_without_path or os.path.basename(item.bin_name)
 
-    return final_result_item, auto_id_comment, is_new
 
+def _match_key(filename: str, checksum: str) -> MatchKey:
+    return filename, checksum or ""
 
-def get_list_by_using_query(sql_query, params, columns, conn, cur):
-    result_rows = ""  # DataFrame
-    cur.execute(sql_query, params)
-    rows = cur.fetchall()
 
-    if rows is not None and len(rows) > 0:
-        result_rows = pd.DataFrame(data=rows, columns=columns)
-    return result_rows
+def _build_deduped_payload(bin_info_list) -> Tuple[List[dict], Dict[MatchKey, str]]:
+    """Deduplicate by filename+checksum; return API payload and key→api_id map."""
+    key_to_id: Dict[MatchKey, str] = {}
+    items_payload: List[dict] = []
 
+    for item in bin_info_list:
+        filename = _item_filename(item)
+        checksum = item.checksum or ""
+        key = _match_key(filename, checksum)
+        if key in key_to_id:
+            continue
+        api_id = str(len(items_payload))
+        key_to_id[key] = api_id
+        items_payload.append({
+            "id": api_id,
+            "filename": filename,
+            "checksum": checksum,
+            "tlsh": item.tlsh or CONST_TLSH_NULL,
+        })
+
+    return items_payload, key_to_id
+
+
+def _apply_match_result_to_item(item, result: Optional[dict]) -> None:
+    if not result or not result.get("matched"):
+        item.set_comment("New Binary/")
+        item.is_new_bin = True
+        return
+
+    oss_rows = result.get("oss_items") or []
+    if not oss_rows:
+        item.set_comment("New Binary/")
+        item.is_new_bin = True
+        return
+
+    item.is_new_bin = False
+    item.set_comment("")
+    for row_idx, row in enumerate(oss_rows):
+        if row_idx == 0:
+            item.set_oss_name(row.get("oss_name") or "")
+            item.set_oss_version(row.get("oss_version") or "")
+            item.set_license(row.get("license") or "")
+        else:
+            item.set_additional_oss_items(
+                f"{row.get('oss_name') or ''}\t"
+                f"{row.get('oss_version') or ''}\t"
+                f"{row.get('license') or ''}"
+            )
+
+
+def get_oss_info_from_db(bin_info_list, kb_url: str = "", kb_token: str = ""):
+    """
+    Call ldb_service /binary/match and apply OSS info.
+    Deduplicates by filename+checksum before the API call and maps results back.
+    """
+    if not bin_info_list:
+        return bin_info_list
+
+    base_url, token = resolve_kb_config(kb_url, kb_token)
+    items_payload, key_to_id = _build_deduped_payload(bin_info_list)
+    if not items_payload:
+        return bin_info_list
+
+    endpoint = f"{base_url.rstrip('/')}{_BINARY_MATCH_PATH}"
+    logger.info(f"Querying KB binary match: {endpoint}")
+
+    results_by_id = {}
+    kb_reachable_logged = False
+    try:
+        for chunk_start in range(0, len(items_payload), _CHUNK_SIZE):
+            chunk = items_payload[chunk_start: chunk_start + _CHUNK_SIZE]
+            response, unreachable = _post_binary_match(base_url, token, chunk)
+            if unreachable:
+                # Host not reachable — do not attempt remaining chunks
+                break
+            if not kb_reachable_logged:
+                logger.info(f"KB({base_url}) reachable")
+                kb_reachable_logged = True
+            if response is None:
+                logger.warning(
+                    f"Binary match chunk failed "
+                    f"({chunk_start}:{chunk_start + len(chunk)}); "
+                    "keeping results so far and continuing with next chunks."
+                )
+                continue
+            for result in response.get("results", []):
+                results_by_id[str(result.get("id"))] = result
+    except Exception as error:
+        logger.warning(f"KB({base_url}) binary match API failed: {error}")
+
+    for item in bin_info_list:
+        try:
+            key = _match_key(_item_filename(item), item.checksum or "")
+            api_id = key_to_id.get(key)
+            if api_id is None or api_id not in results_by_id:
+                continue
+            _apply_match_result_to_item(item, results_by_id[api_id])
+        except Exception as error:
+            logger.warning(f"READ OSS :{error}")
+
+    return bin_info_list
+
+
+def _post_binary_match(kb_url: str, kb_token: str, items: list) -> PostMatchResult:
+    """POST one chunk. Returns (body, unreachable). unreachable stops further chunks."""
+    data = json.dumps({"items": items}).encode("utf-8")
+    request = urllib.request.Request(
+        f"{kb_url.rstrip('/')}{_BINARY_MATCH_PATH}",
+        data=data,
+        method="POST",
+    )
+    request.add_header("Accept", "application/json")
+    request.add_header("Content-Type", "application/json")
+    if kb_token:
+        request.add_header("Authorization", f"Bearer {kb_token}")
 
-def disconnect_lge_bin_db(conn, cur):
-    # Close connection
     try:
-        cur.close()
-        conn.close()
-    except Exception:
-        pass
+        with urllib.request.urlopen(request, timeout=_HTTP_TIMEOUT_SEC) as response:
+            body = response.read().decode()
+            return (json.loads(body) if body else {}), False
+    except urllib.error.HTTPError as ex:
+        body = ""
+        try:
+            body = ex.read().decode()
+        except Exception:
+            pass
+        # Host responded → reachable; caller may continue with next chunks
+        logger.warning(
+            f"KB({kb_url}) reachable but binary match HTTP {ex.code}: {body or ex.reason}"
+        )
+        return None, False
+    except urllib.error.URLError as ex:
+        logger.warning(f"KB({kb_url}) Unreachable: {ex.reason if hasattr(ex, 'reason') else ex}")
+        return None, True
+    except Exception as ex:
+        logger.warning(f"KB({kb_url}) binary match failed: {ex}")
+        return None, False

src/fosslight_android/android_binary_analysis.py

@@ -24,7 +24,6 @@
 import multiprocessing
 import parmap
 import numpy as np
-from functools import partial
 from fosslight_util.set_log import init_log
 from ._util import (
     read_file,
@@ -665,9 +664,7 @@ def set_checksum_tlsh_and_get_oss_from_db_after_remove_duplication(remove_list_f
     final_bin_info = return_list[:]
     remove_duplicated_binaries_by_checking_checksum(remove_list_file)
 
-    func = partial(get_oss_info_from_db, platform_version)
-    return_oss_list = do_multi_process(func, final_bin_info)
-    final_bin_info = return_oss_list[:]
+    final_bin_info = get_oss_info_from_db(final_bin_info)
 
 
 def get_checksum_tlsh(bin_info_list, return_bin_list):