Kangjie Chen - RankMe

Year	Rank	Type	Title / Venue / Authors
2026	J	`jnl`	DECEIVE-AFC: Adversarial Claim Attacks against Search-Enabled LLM-based Fact-Checking Systems. CoRR Haoran Ou, Kangjie Chen, Gelei Deng, Hangcheng Liu, Jie Zhang, Tianwei Zhang, Kwok-Yan Lam
2026	J	`jnl`	SafeRedir: Prompt Embedding Redirection for Robust Unlearning in Image Generation Models. CoRR Renyang Liu, Kangjie Chen, Han Qiu, Jie Zhang, Kwok-Yan Lam, Tianwei Zhang, See-Kiong Ng
2026	J	`jnl`	X-World: Controllable Ego-Centric Multi-Camera World Models for Scalable End-to-End Driving. CoRR Chaoda Zheng, Sean Li, Jinhao Deng, Zhennan Wang, Shijia Chen, Liqiang Xiao, Ziheng Chi, Hongbin Lin, Kangjie Chen, Boyang Wang, Yu Zhang, Xianming Liu
2025	J	`jnl`	BadLingual: A Novel Lingual-Backdoor Attack against Large Language Models. CoRR Zihan Wang, Hongwei Li, Rui Zhang, Wenbo Jiang, Kangjie Chen, Tianwei Zhang, Qingchuan Zhao, Guowen Xu
2025	J	`jnl`	CREST-Search: Comprehensive Red-teaming for Evaluating Safety Threats in Large Language Models Powered by Web Search. CoRR Haoran Ou, Kangjie Chen, Xingshuo Han, Gelei Deng, Jie Zhang, Han Qiu, Tianwei Zhang
2025	J	`jnl`	Coward: Toward Practical Proactive Federated Backdoor Defense via Collision-based Watermark. CoRR Wenjie Li, Siying Gu, Yiming Li, Kangjie Chen, Zhili Chen, Tianwei Zhang, Shu-Tao Xia, Dacheng Tao
2025	A*	`conf`	HRAvatar: High-Quality and Relightable Gaussian Head Avatar. CVPR Dongbin Zhang, Yunfei Liu, Lijian Lin, Ye Zhu, Kangjie Chen, Minghan Qin, Yu Li, Haoqian Wang
2025	J	`jnl`	HRAvatar: High-Quality and Relightable Gaussian Head Avatar. CoRR Dongbin Zhang, Yunfei Liu, Lijian Lin, Ye Zhu, Kangjie Chen, Minghan Qin, Yu Li, Haoqian Wang
2025	J	`jnl`	Impact-driven Context Filtering For Cross-file Code Completion. CoRR Yanzhou Li, Shangqing Liu, Kangjie Chen, Tianwei Zhang, Yang Liu
2025	J	`jnl`	Picky LLMs and Unreliable RMs: An Empirical Study on Safety Alignment after Instruction Tuning. CoRR Guanlin Li, Kangjie Chen, Shangwei Guo, Jie Zhang, Han Qiu, Chao Zhang, Guoyin Wang, Tianwei Zhang, Jiwei Li
2025	J	`jnl`	Quantifying and Alleviating Co-Adaptation in Sparse-View 3D Gaussian Splatting. CoRR Kangjie Chen, Yingji Zhong, Zhihao Li, Jiaqi Lin, Youyu Chen, Minghan Qin, Haoqian Wang
2025	A*	`conf`	SLGaussian: Fast Language Gaussian Splatting in Sparse Views. ACM Multimedia Kangjie Chen, Bingquan Dai, Minghan Qin, Dongbin Zhang, Peihao Li, Yingshuang Zou, Haoqian Wang
2025	J	`jnl`	TEAR: Temporal-aware Automated Red-teaming for Text-to-Video Models. CoRR Jiaming He, Guanyu Hou, Hongwei Li, Zhicong Huang, Kangjie Chen, Yi Yu, Wenbo Jiang, Guowen Xu, Tianwei Zhang
2025	A*	`conf`	TRUST-VLM: Thorough Red-Teaming for Uncovering Safety Threats in Vision-Language Models. ICML Kangjie Chen, Muyang Li, Guanlin Li, Shudong Zhang, Shangwei Guo, Tianwei Zhang
2025	J	`jnl`	Towards Effective Prompt Stealing Attack against Text-to-Image Diffusion Models. CoRR Shiqian Zhao, Chong Wang, Yiming Li, Yihao Huang, Wenjie Qu, Siew-Kei Lam, Yi Xie, Kangjie Chen, Jie Zhang, Tianwei Zhang
2025	A*	`conf`	USD: NSFW Content Detection for Text-to-Image Models via Scene Graph. USENIX Security Symposium Yuyang Zhang, Kangjie Chen, Xudong Jiang, Jiahui Wen, Yihui Jin, Ziyou Liang, Yihao Huang, Run Wang, Lina Wang
2025	J	`jnl`	Unmasking Backdoors: An Explainable Defense via Gradient-Attention Anomaly Scoring for Pre-trained Language Models. CoRR Anindya Sundar Das, Kangjie Chen, Monowar Bhuyan
2024	A*	`conf`	ART: Automatic Red-teaming for Text-to-Image Models to Protect Benign Users. NeurIPS Guanlin Li, Kangjie Chen, Shudong Zhang, Jie Zhang, Tianwei Zhang
2024	J	`jnl`	ART: Automatic Red-teaming for Text-to-Image Models to Protect Benign Users. CoRR Guanlin Li, Kangjie Chen, Shudong Zhang, Jie Zhang, Tianwei Zhang
2024	A*	`conf`	BadEdit: Backdooring Large Language Models by Model Editing. ICLR Yanzhou Li, Tianlin Li, Kangjie Chen, Jian Zhang, Shangqing Liu, Wenhan Wang, Tianwei Zhang, Yang Liu
2024	J	`jnl`	BadEdit: Backdooring large language models by model editing. CoRR Yanzhou Li, Tianlin Li, Kangjie Chen, Jian Zhang, Shangqing Liu, Wenhan Wang, Tianwei Zhang, Yang Liu
2024	J	`jnl`	Boosting Black-Box Attack to Deep Neural Networks With Conditional Diffusion Models. IEEE Trans. Inf. Forensics Secur. Renyang Liu, Wei Zhou, Tianwei Zhang, Kangjie Chen, Jun Zhao, Kwok-Yan Lam
2024	A*	`conf`	EvilEdit: Backdooring Text-to-Image Diffusion Models in One Second. ACM Multimedia Hao Wang, Shangwei Guo, Jialing He, Kangjie Chen, Shudong Zhang, Tianwei Zhang, Tao Xiang
2024	J	`jnl`	MIP: CLIP-based Image Reconstruction from PEFT Gradients. CoRR Peiheng Zhou, Ming Hu, Xiaofei Xie, Yihao Huang, Kangjie Chen, Mingsong Chen
2024	A	`conf`	Protecting Confidential Virtual Machines from Hardware Performance Counter Side Channels. DSN Xiaoxuan Lou, Kangjie Chen, Guowen Xu, Han Qiu, Shangwei Guo, Tianwei Zhang
2024	J	`jnl`	SLGaussian: Fast Language Gaussian Splatting in Sparse Views. CoRR Kangjie Chen, Bingquan Dai, Minghan Qin, Dongbin Zhang, Peihao Li, Yingshuang Zou, Haoqian Wang
2024	J	`jnl`	Towards Action Hijacking of Large Language Model-based Agent. CoRR Yuyang Zhang, Kangjie Chen, Xudong Jiang, Yuxiang Sun, Run Wang, Lina Wang
2023	J	`jnl`	ADS-Lead: Lifelong Anomaly Detection in Autonomous Driving Systems. IEEE Trans. Intell. Transp. Syst. Xingshuo Han, Yuan Zhou, Kangjie Chen, Han Qiu, Meikang Qiu, Yang Liu, Tianwei Zhang
2023	J	`jnl`	Boosting Black-box Attack to Deep Neural Networks with Conditional Diffusion Models. CoRR Renyang Liu, Wei Zhou, Tianwei Zhang, Kangjie Chen, Jun Zhao, Kwok-Yan Lam
2023	A*	`conf`	Clean-image Backdoor: Attacking Multi-label Models with Poisoned Labels Only. ICLR Kangjie Chen, Xiaoxuan Lou, Guowen Xu, Jiwei Li, Tianwei Zhang
2023	J	`jnl`	Extracting Cloud-based Model with Prior Knowledge. CoRR Shiqian Zhao, Kangjie Chen, Meng Hao, Jian Zhang, Guowen Xu, Hongwei Li, Tianwei Zhang
2023	A*	`conf`	GuardHFL: Privacy Guardian for Heterogeneous Federated Learning. ICML Hanxiao Chen, Meng Hao, Hongwei Li, Kangjie Chen, Guowen Xu, Tianwei Zhang, Xilin Zhang
2023	—	`conf`	Multi-target Backdoor Attacks for Code Pre-trained Models. ACL (1) Yanzhou Li, Shangqing Liu, Kangjie Chen, Xiaofei Xie, Tianwei Zhang, Yang Liu
2023	J	`jnl`	Multi-target Backdoor Attacks for Code Pre-trained Models. CoRR Yanzhou Li, Shangqing Liu, Kangjie Chen, Xiaofei Xie, Tianwei Zhang, Yang Liu
2023	J	`jnl`	Omnipotent Adversarial Training for Unknown Label-noisy and Imbalanced Datasets. CoRR Guanlin Li, Kangjie Chen, Yuan Xu, Han Qiu, Tianwei Zhang
2022	A*	`conf`	BadPre: Task-agnostic Backdoor Attacks to Pre-trained NLP Foundation Models. ICLR Kangjie Chen, Yuxian Meng, Xiaofei Sun, Shangwei Guo, Tianwei Zhang, Jiwei Li, Chun Fan
2022	J	`jnl`	ShiftNAS: Towards Automatic Generation of Advanced Mulitplication-Less Neural Networks. CoRR Xiaoxuan Lou, Guowen Xu, Kangjie Chen, Guanlin Li, Jiwei Li, Tianwei Zhang
2021	—	`conf`	A Unified Anomaly Detection Methodology for Lane-Following of Autonomous Driving Systems. ISPA/BDCloud/SocialCom/SustainCom Xingshuo Han, Kangjie Chen, Yuan Zhou, Meikang Qiu, Chun Fan, Yang Liu, Tianwei Zhang
2021	J	`jnl`	BadPre: Task-agnostic Backdoor Attacks to Pre-trained NLP Foundation Models. CoRR Kangjie Chen, Yuxian Meng, Xiaofei Sun, Shangwei Guo, Tianwei Zhang, Jiwei Li, Chun Fan
2021	A	`conf`	Stealing Deep Reinforcement Learning Models for Fun and Profit. AsiaCCS Kangjie Chen, Shangwei Guo, Tianwei Zhang, Xiaofei Xie, Yang Liu
2021	A	`conf`	Temporal Watermarks for Deep Reinforcement Learning Models. AAMAS Kangjie Chen, Shangwei Guo, Tianwei Zhang, Shuxin Li, Yang Liu
2021	J	`jnl`	Vulnerability Assessment of Deep Reinforcement Learning Models for Power System Topology Optimization. IEEE Trans. Smart Grid Yan Zheng, Ziming Yan, Kangjie Chen, Jianwen Sun, Yan Xu, Yang Liu
2020	J	`jnl`	Stealing Deep Reinforcement Learning Models for Fun and Profit. CoRR Kangjie Chen, Tianwei Zhang, Xiaofei Xie, Yang Liu
2020	A*	`conf`	Stealthy and Efficient Adversarial Attacks against Deep Reinforcement Learning. AAAI Jianwen Sun, Tianwei Zhang, Xiaofei Xie, Lei Ma, Yan Zheng, Kangjie Chen, Yang Liu
2020	J	`jnl`	Stealthy and Efficient Adversarial Attacks against Deep Reinforcement Learning. CoRR Jianwen Sun, Tianwei Zhang, Xiaofei Xie, Lei Ma, Yan Zheng, Kangjie Chen, Yang Liu
2017	A*	`conf`	Defending Against Man-In-The-Middle Attack in Repeated Games. IJCAI Shuxin Li, Xiaohong Li, Jianye Hao, Bo An, Zhiyong Feng, Kangjie Chen, Chengwei Zhang

import hashlib
import inspect
import json
import os
import time
import threading
from datetime import datetime, timezone
from typing import Dict, Any, Optional

from redb.extractors.enum import Tag
from redb.extractors.extractor import Extractor
import magic
import pefile
import ppdeep
import tlsh

# Import our BinjaDecompiler (conditional)
from redb.extractors.decompiler.bninja.decompiler import BinaryNinjaDecompiler

class DecompileBinja(Extractor):
    def __init__(
        self,
        filepath,
        log,
        exporters=None,
        index_prefix=None,
        elastic_index=None,
        known_benign=False,
        known_malicious=False,
        filetype=None,
    ):
        super().__init__(
            filepath,
            log,
            exporters,
            index_prefix,
            elastic_index,
            known_benign,
            known_malicious,
        )
        self.log.debug(inspect.currentframe().f_code.co_name)

        # Check if Binary Ninja is available
       # if not BINARYNINJA_AVAILABLE:
        #     self.log.error("Binary Ninja is not available in this container")
        #    raise ImportError(
        #        "Binary Ninja module not found - not available in feature extraction container"
        #    )
        self.analysis_results = None
        self.binja_decompiler = None
        self.filetype = filetype

        # Convert TIMEOUT to integer with a default of 1200 seconds (20 minutes)
        try:
            self.BINJA_TIMEOUT = int(os.getenv("BINJA_TIMEOUT", "1200"))
        except ValueError:
            self.log.warning(
                "Invalid BINJA_TIMEOUT value, using default of 1200 seconds"
            )
            self.BINJA_TIMEOUT = 1200

        # Convert TIMEOUT to integer with a default of 1200 seconds (20 minutes)
        try:
            self.DECOMPILE_EXTRACTOR_TIMEOUT = int(
                os.getenv("DECOMPILE_EXTRACTOR_TIMEOUT", "2580")
            )
        except ValueError:
            self.log.warning(
                "Invalid DECOMPILE_EXTRACTOR_TIMEOUT value, using default of 2580 seconds"
            )
            self.DECOMPILE_EXTRACTOR_TIMEOUT = 2580

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.cleanup_run()

    def calculate_md5(self, input_str):
        """Calculate MD5 hash of a string."""
        return hashlib.md5(input_str.encode("utf-8")).hexdigest()

    def is_dotnet(self):
        """Check if the binary is a .NET assembly.

        Returns:
            bool: True if the file is a .NET assembly, False otherwise
        """
        try:
            if self.filetype == "pebin":
                file_type = magic.from_buffer(self.binary)
                if ".Net" in file_type:
                    return True
                pe = pefile.PE(self.filepath)
                for entry in pe.OPTIONAL_HEADER.DATA_DIRECTORY:
                    # IMAGE_DIRECTORY_ENTRY_COM_DESCRIPTOR is typically 14
                    if (
                        entry.name == "IMAGE_DIRECTORY_ENTRY_COM_DESCRIPTOR"
                        and entry.Size > 0
                    ):
                        return True
                return False
        except AttributeError as e:
            self.log.error(
                f"AttributeError error dotnet file {self.hash.sha256} Full error : {e}"
            )
            return False

    def cleanup_run(self):
        """Clean up after analysis."""
        try:
            # BinaryNinjaDecompiler uses context manager pattern (__enter__/__exit__)
            # Cleanup happens automatically when exiting the 'with' block
            self.binja_decompiler = None

            # Force garbage collection
            import gc

            gc.collect()

        except Exception as e:
            self.log.error(f"Error in cleanup: {e}")

    def analyze_binary(self) -> Optional[Dict[str, Any]]:
        """Run Binary Ninja analysis and return results."""
        self.log.debug("Starting binary analysis")

        try:
            # Use BinaryNinjaDecompiler as a context manager to ensure proper setup/cleanup
            with BinaryNinjaDecompiler(
                filepath=self.filepath,
                timeout=self.BINJA_TIMEOUT,
                log=self.log,
                exporters=self.exporters,
                index_prefix=self.index_prefix,
                filetype=self.filetype,
            ) as decompiler:
                self.binja_decompiler = decompiler

                if decompiler.extract():
                    # Store results before context manager exits
                    results = decompiler.analysis_results
                    return results
                else:
                    self.log.error("BinaryNinjaDecompiler extraction failed")
                    return None

        except Exception as e:
            self.log.error(f"Error in Binary Ninja analysis: {e}")
            import traceback
            self.log.error(f"Traceback: {traceback.format_exc()}")
            return None

        finally:
            self.cleanup_run()

    def extract(self):
        """Extract and process all analysis results."""
        self.log.debug(inspect.currentframe().f_code.co_name)

        # Create a flag to track if extraction completed
        extraction_completed = False
        extraction_result = False
        extraction_error = None

        # Define the extraction process as a separate function
        def do_extraction():
            nonlocal extraction_completed, extraction_result, extraction_error
            try:
                results = self.analyze_binary()
                if not results:
                    extraction_result = False
                else:
                    self.analysis_results = results
                    extraction_result = True
            except Exception as e:
                extraction_error = e
                extraction_result = False
            finally:
                extraction_completed = True

        # Start extraction in a separate thread
        extraction_thread = threading.Thread(target=do_extraction)
        extraction_thread.daemon = True
        extraction_thread.start()

        # Wait for the extraction to complete or timeout
        start_time = time.time()
        while (
            not extraction_completed
            and (time.time() - start_time) < self.DECOMPILE_EXTRACTOR_TIMEOUT
        ):
            time.sleep(1)

        if not extraction_completed:
            self.log.error(
                f"Extraction timed out after {self.DECOMPILE_EXTRACTOR_TIMEOUT} seconds"
            )
            # Force cleanup
            self.cleanup_run()
            return None

        if extraction_error:
            self.log.error(f"Error in extraction: {extraction_error}")
            return None

        # Return the actual analysis results, not just a boolean
        return self.analysis_results if extraction_result else None

    def prepare_export_data(self, exporter_type: str) -> Any:
        """Prepare data for database export."""
        self.log.debug(inspect.currentframe().f_code.co_name)
        if not self.analysis_results:
            return None

        # # Delegate to the BinjaDecompiler for consistent export formatting
        # if self.binja_decompiler:
        #     return self.binja_decompiler.prepare_export_data(exporter_type)
        # else:
        #     self.log.error("BinjaDecompiler not available for export preparation")
        #     return None

        if exporter_type == "ClickHouseExporter":
            now = datetime.now(timezone.utc)

            def prepare_array_field(value, array_type):
                """Helper to prepare array fields with proper null handling"""
                if value is None:
                    return []
                return value

            # Add a helper function to handle empty strings
            def ensure_not_empty(value, default="UNKNOWN"):
                """Ensure a string value is not empty"""
                if value is None or value == "":
                    return default
                return value

            def ssdeep_disassembly(func):
                try:
                    if len(func) > 1:
                        return ppdeep.hash(func)
                    return ""
                except Exception as e:
                    self.log.error(f"Error in disassembly ssdeep hash calculation: {e}")
                    return ""

            def tlsh_disassembly(func):
                try:
                    if len(func) >= 50:
                        return tlsh.hash(func.encode("utf-8"))
                    return ""
                except Exception as e:
                    self.log.error(f"Error in disassembly tlsh hash calculation: {e}")
                    return ""

            return {
                "multi_table": True,
                "decompiled_content": {
                    "table": "code_binja_decompiled_functions_content",
                    "data": [
                        [
                            f["decompiled_function_hash"],
                            f["decompiled_function"],
                            f["function_type"],
                            now,
                        ]
                        for f in self.analysis_results["decompiled"]
                    ],
                    "column_names": [
                        "decompiled_function_hash",
                        "decompiled_function",
                        "function_type",
                        "analysis_date",
                    ],
                    "column_type_names": [
                        "FixedString(64)",
                        "String",
                        "Enum8('USER'=1, 'LIBRARY'=2, 'THUNK'=3, 'EXTERNAL'=4, 'UNKNOWN'=5)",
                        "DateTime64(3, 'UTC')",
                    ],
                },
                "decompiled_refs": {
                    "table": "code_binja_decompiled_functions_references",
                    "data": [
                        [
                            self.analysis_results["sha256"],
                            self.analysis_results["sha1"],
                            self.analysis_results["md5"],
                            f["decompiled_function_hash"],
                            f["disassembled_function_hash"],  # New linking field
                            f["decompiled_function_name"],
                            f["decompiled_function_prototype"],
                            f["decompiled_function_address"],
                            now,
                        ]
                        for f in self.analysis_results["decompiled"]
                    ],
                    "column_names": [
                        "sha256",
                        "sha1",
                        "md5",
                        "decompiled_function_hash",
                        "disassembled_function_hash",  # New linking field
                        "decompiled_function_name",
                        "decompiled_function_prototype",
                        "decompiled_function_address",
                        "analysis_date",
                    ],
                    "column_type_names": [
                        "FixedString(64)",
                        "FixedString(40)",
                        "FixedString(32)",
                        "FixedString(64)",
                        "Nullable(FixedString(64))",  # New linking field
                        "LowCardinality(String)",
                        "LowCardinality(String)",
                        "UInt64",
                        "DateTime64(3, 'UTC')",
                    ],
                },
                "disassembled_content": {
                    "table": "code_binja_disassembled_functions_content",
                    "data": [
                        [
                            f["disassembled_function_hash"],
                            f.get("disassembled_function", ""),
                            f.get("disassembled_function_no_addresses", ""),
                            f.get("function_type", "UNKNOWN"),
                            f.get("instructions_count", 0),
                            prepare_array_field(
                                f.get("instructions_types"), "LowCardinality(String)"
                            ),
                            f.get("control_flow_count", 0),
                            prepare_array_field(
                                f.get("memory_access_pattern"), "LowCardinality(String)"
                            ),
                            prepare_array_field(
                                f.get("register_usage"), "LowCardinality(String)"
                            ),
                            f.get("data_references_count", 0),
                            f.get("max_block_size", 0),
                            f.get("num_calls", 0),
                            f.get("stack_size", 0),
                            now,
                        ]
                        for f in self.analysis_results["disassembled"]
                    ],
                    "column_names": [
                        "disassembled_function_hash",
                        "disassembled_function",
                        "disassembled_function_no_addresses",
                        "function_type",
                        "instructions_count",
                        "instructions_types",
                        "control_flow_count",
                        "memory_access_pattern",
                        "register_usage",
                        "data_references_count",
                        "max_block_size",
                        "num_calls",
                        "stack_size",
                        "analysis_date",
                    ],
                    "column_type_names": [
                        "FixedString(64)",
                        "String",
                        "String",
                        "Enum8('USER'=1, 'LIBRARY'=2, 'THUNK'=3, 'EXTERNAL'=4, 'UNKNOWN'=5)",
                        "UInt32",
                        "Array(LowCardinality(String))",
                        "UInt32",
                        "Array(LowCardinality(String))",
                        "Array(LowCardinality(String))",
                        "UInt32",
                        "Nullable(UInt32)",
                        "Nullable(UInt32)",
                        "Nullable(Int32)",
                        "DateTime64(3, 'UTC')",
                    ],
                },
                "disassembled_refs": {
                    "table": "code_binja_disassembled_functions_references",
                    "data": [
                        [
                            self.analysis_results["sha256"],
                            self.analysis_results["sha1"],
                            self.analysis_results["md5"],
                            f["disassembled_function_hash"],
                            f["decompiled_function_hash"],
                            f["disassembled_function_name"],
                            f["disassembled_function_address"],
                            ssdeep_disassembly(
                                f.get("disassembled_function_no_addresses", "")
                            ),
                            tlsh_disassembly(
                                f.get("disassembled_function_no_addresses", "")
                            ),
                            now,
                        ]
                        for f in self.analysis_results["disassembled"]
                    ],
                    "column_names": [
                        "sha256",
                        "sha1",
                        "md5",
                        "disassembled_function_hash",
                        "decompiled_function_hash",
                        "disassembled_function_name",
                        "disassembled_function_address",
                        "ssdeep_disassembly",
                        "tlsh_disassembly",
                        "analysis_date",
                    ],
                    "column_type_names": [
                        "FixedString(64)",
                        "FixedString(40)",
                        "FixedString(32)",
                        "FixedString(64)",
                        "Nullable(FixedString(64))",
                        "LowCardinality(String)",
                        "UInt64",
                        "Nullable(String)",
                        "Nullable(FixedString(72))",
                        "DateTime64(3, 'UTC')",
                    ],
                },
                "cfg_blocks": {
                    "table": "code_binja_cfg_blocks",
                    "data": [
                        [
                            b["block_id"],
                            self.analysis_results["sha256"],
                            self.analysis_results["sha1"],
                            self.analysis_results["md5"],
                            b["function_address"],
                            b["block_start_address"],
                            b["block_end_address"],
                            b["block_size"],
                            b["instructions_count"],
                            b["block_instructions"],  # MD5 hash of instructions
                            b["predecessor_blocks"],
                            b["successor_blocks"],
                            b["depth"],
                            b["position"],
                            b["branch_type"],
                            b["block_type"],
                            b["flags"],
                            b["dominators"],
                            b["post_dominators"],
                            now,
                        ]
                        # Flatten: iterate through all functions, then all blocks in each function
                        for func_cfg in self.analysis_results["cfg"]
                        if func_cfg is not None  # Handle None from failed extractions
                        for b in func_cfg["blocks"]
                    ],
                    "column_names": [
                        "block_id",
                        "sha256",
                        "sha1",
                        "md5",
                        "function_address",
                        "block_start_address",
                        "block_end_address",
                        "block_size",
                        "instructions_count",
                        "block_instructions_hash",
                        "predecessor_blocks",
                        "successor_blocks",
                        "depth",
                        "position",
                        "branch_type",
                        "block_type",
                        "flags",
                        "dominators",
                        "post_dominators",
                        "analysis_date",
                    ],
                    "column_type_names": [
                        "FixedString(64)",  # block_id (SHA256)
                        "FixedString(64)",  # sha256
                        "FixedString(40)",  # sha1
                        "FixedString(32)",  # md5
                        "UInt64",  # function_address
                        "UInt64",  # block_start_address
                        "UInt64",  # block_end_address
                        "UInt32",  # block_size
                        "UInt16",  # instructions_count
                        "FixedString(32)",  # block_instructions_hash (MD5)
                        "Array(UInt64)",  # predecessor_blocks
                        "Array(UInt64)",  # successor_blocks
                        "UInt16",  # depth
                        "UInt16",  # position
                        "Enum8('DIRECT'=1, 'CONDITIONAL'=2, 'CALL'=3, 'RETURN'=4, 'FALLTHROUGH'=5, 'INDIRECT'=6, 'UNKNOWN'=7)",  # branch_type
                        "Enum8('CODE'=1, 'DATA'=2, 'THUNK'=3)",  # block_type
                        "Array(String)",  # flags (EntryBlock, ExitBlock, LoopBlock)
                        "Array(UInt16)",  # dominators
                        "Array(UInt16)",  # post_dominators
                        "DateTime64(3, 'UTC')",  # analysis_date
                    ],
                },
                "function_analysis_errors": {
                    "table": "function_analysis_errors_binja",
                    "data": [
                        [
                            self.analysis_results["sha256"],
                            f["function_name"],
                            f["function_address"],
                            f.get("error_location", "unknown"),
                            f.get("error_message", ""),
                            f.get("error_details", ""),
                            f.get("error_type", "unknown"),
                            self.calculate_md5(
                                f"{f['error_message']}{f['function_name']}{f['function_address']}{f['error_location']}"
                            ),
                            "new",
                            now,
                        ]
                        for f in self.analysis_results["errors"]
                    ],
                    "column_names": [
                        "sha256",
                        "function_name",
                        "function_address",
                        "error_location",
                        "error_message",
                        "error_details",
                        "error_type",
                        "error_hash",
                        "status",
                        "analysis_date",
                    ],
                    "column_type_names": [
                        "FixedString(64)",
                        "Nullable(String)",
                        "UInt64",
                        "LowCardinality(String)",
                        "Nullable(String)",
                        "Nullable(String)",
                        "Nullable(String)",
                        "FixedString(32)",
                        "Enum8('new'=1, 'investigating'=2, 'fixed'=3, 'wontfix'=4)",
                        "DateTime64(3, 'UTC')",
                    ],
                },
            }

    def tag(self) -> str:
        """Return the tag for this extractor."""
        return Tag.DECOMPILED.value

    def get_clickhouse_table(self) -> str:
        """Not used directly as we're handling multiple tables."""
        pass


if __name__ == "__main__":
    # Setup basic logging
    import logging

    logging.basicConfig(level=logging.INFO)
    logger = logging.getLogger("DecompileBinja")

    # Parse command line arguments
    import argparse

    parser = argparse.ArgumentParser(description="Binary Ninja Decompiler Wrapper")
    parser.add_argument("filepath", help="Path to the binary file to analyze")
    parser.add_argument(
        "--output", "-o", help="Output JSON file path (default: stdout)"
    )
    parser.add_argument(
        "--timeout",
        "-t",
        type=int,
        default=1200,
        help="Analysis timeout in seconds (default: 1200)",
    )
    args = parser.parse_args()

    # Create and run the extractor
    with DecompileBinja(args.filepath, logger) as extractor:
        success = extractor.extract()

        if not success:
            logger.error("Analysis failed")
            exit(1)

        # Output results
        if args.output:
            with open(args.output, "w") as f:
                json.dump(extractor.analysis_results, f)
            logger.info(f"Results written to {args.output}")
        else:
            print(json.dumps(extractor.analysis_results))