diff --git a/pyproject.toml b/pyproject.toml index c091654..bafc874 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,7 +32,6 @@ dependencies = [ "numpy", "pandas", "parmap", - "psycopg2-binary>=2.9.10", "python-dateutil", "py-tlsh", "pytz", diff --git a/src/fosslight_android/_binary_db_controller.py b/src/fosslight_android/_binary_db_controller.py index 1bb7c1c..123b059 100644 --- a/src/fosslight_android/_binary_db_controller.py +++ b/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 diff --git a/src/fosslight_android/android_binary_analysis.py b/src/fosslight_android/android_binary_analysis.py index 485875d..04166f4 100755 --- a/src/fosslight_android/android_binary_analysis.py +++ b/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):