more work on v2 java bridge

This commit is contained in:
Alexander Kiselev
2026-02-04 14:14:15 -08:00
parent ac1fb7b931
commit a550ad1d35
36 changed files with 1777 additions and 5277 deletions
+6 -6
View File
@@ -26,11 +26,6 @@ jobs:
distribution: 'temurin'
java-version: '17'
- name: Install Python 3
uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Install Rust toolchain
uses: dtolnay/rust-toolchain@stable
@@ -44,11 +39,14 @@ jobs:
~/.local/share/ghidra-cli
~/Library/Application Support/ghidra-cli
~/AppData/Local/ghidra-cli
key: ghidra-${{ matrix.os }}-v1
key: ghidra-${{ matrix.os }}-v2
- name: Build
run: cargo build --verbose
- name: Build test fixture
run: rustc --edition 2021 -o tests/fixtures/sample_binary tests/fixtures/sample_binary.rs
- name: Setup Ghidra
run: cargo run -- setup --force
@@ -57,6 +55,8 @@ jobs:
- name: Run integration tests
run: cargo test --test '*' --verbose
env:
RUST_LOG: info
build:
name: Build ${{ matrix.target }}
+6 -6
View File
@@ -26,11 +26,6 @@ jobs:
distribution: 'temurin'
java-version: '17'
- name: Install Python 3
uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Install Rust toolchain
uses: dtolnay/rust-toolchain@stable
@@ -44,11 +39,14 @@ jobs:
~/.local/share/ghidra-cli
~/Library/Application Support/ghidra-cli
~/AppData/Local/ghidra-cli
key: ghidra-${{ matrix.os }}-v1
key: ghidra-${{ matrix.os }}-v2
- name: Build
run: cargo build --verbose
- name: Build test fixture
run: rustc --edition 2021 -o tests/fixtures/sample_binary tests/fixtures/sample_binary.rs
- name: Setup Ghidra
run: cargo run -- setup --force
@@ -57,3 +55,5 @@ jobs:
- name: Run integration tests
run: cargo test --test '*' --verbose
env:
RUST_LOG: info
+4 -18
View File
@@ -22,28 +22,14 @@ pub struct BridgeConfig {
/// Ensure a bridge is running for the given project.
/// If import mode, starts with the binary. If process mode, opens existing program.
/// Returns the port number for connecting.
pub fn ensure_bridge(
config: &BridgeConfig,
mode: BridgeStartMode,
) -> Result<u16> {
bridge::ensure_bridge_running(
&config.project_path,
&config.ghidra_install_dir,
mode,
)
pub fn ensure_bridge(config: &BridgeConfig, mode: BridgeStartMode) -> Result<u16> {
bridge::ensure_bridge_running(&config.project_path, &config.ghidra_install_dir, mode)
}
/// Start a new bridge for the given project.
/// Returns the port number for connecting.
pub fn start_bridge(
config: &BridgeConfig,
mode: BridgeStartMode,
) -> Result<u16> {
bridge::start_bridge(
&config.project_path,
&config.ghidra_install_dir,
mode,
)
pub fn start_bridge(config: &BridgeConfig, mode: BridgeStartMode) -> Result<u16> {
bridge::start_bridge(&config.project_path, &config.ghidra_install_dir, mode)
}
/// Stop the bridge for a project.
+20 -35
View File
@@ -34,13 +34,9 @@ struct BridgeRequest {
/// How to start the bridge - import a new binary or open an existing program.
pub enum BridgeStartMode {
/// Import a binary file into the project, then start bridge
Import {
binary_path: String,
},
Import { binary_path: String },
/// Open an existing program in the project
Process {
program_name: String,
},
Process { program_name: String },
}
/// Embedded Java bridge script
@@ -84,7 +80,9 @@ pub fn read_port_file(project_path: &Path) -> Result<Option<u16>> {
return Ok(None);
}
let content = std::fs::read_to_string(&path)?;
let port: u16 = content.trim().parse()
let port: u16 = content
.trim()
.parse()
.context("Invalid port in port file")?;
Ok(Some(port))
}
@@ -96,8 +94,7 @@ pub fn read_pid_file(project_path: &Path) -> Result<Option<u32>> {
return Ok(None);
}
let content = std::fs::read_to_string(&path)?;
let pid: u32 = content.trim().parse()
.context("Invalid PID in PID file")?;
let pid: u32 = content.trim().parse().context("Invalid PID in PID file")?;
Ok(Some(pid))
}
@@ -209,16 +206,13 @@ pub fn start_bridge(
let mut cmd = Command::new(&headless_script);
// analyzeHeadless expects: <parent_directory> <project_name>
let ghidra_project_dir = project_path
.parent()
.unwrap_or(project_path);
let ghidra_project_dir = project_path.parent().unwrap_or(project_path);
let ghidra_project_name = project_path
.file_name()
.map(|n| n.to_string_lossy().to_string())
.unwrap_or_else(|| "project".to_string());
cmd.arg(ghidra_project_dir)
.arg(&ghidra_project_name);
cmd.arg(ghidra_project_dir).arg(&ghidra_project_name);
// Add mode-specific args
match &mode {
@@ -226,9 +220,7 @@ pub fn start_bridge(
cmd.arg("-import").arg(binary_path);
}
BridgeStartMode::Process { program_name } => {
cmd.arg("-process")
.arg(program_name)
.arg("-noanalysis");
cmd.arg("-process").arg(program_name).arg("-noanalysis");
}
}
@@ -330,8 +322,8 @@ pub fn send_command(
command: &str,
args: Option<serde_json::Value>,
) -> Result<serde_json::Value> {
let mut stream = TcpStream::connect(format!("127.0.0.1:{}", port))
.context("Failed to connect to bridge")?;
let mut stream =
TcpStream::connect(format!("127.0.0.1:{}", port)).context("Failed to connect to bridge")?;
stream.set_read_timeout(Some(Duration::from_secs(300))).ok();
stream.set_write_timeout(Some(Duration::from_secs(30))).ok();
@@ -355,19 +347,15 @@ pub fn send_command(
let response: BridgeResponse<serde_json::Value> = serde_json::from_str(&response_line)?;
match response.status.as_str() {
"success" => {
Ok(response.data.unwrap_or(serde_json::json!({})))
}
"success" => Ok(response.data.unwrap_or(serde_json::json!({}))),
"error" => {
let msg = response.message.unwrap_or_else(|| "Unknown error".to_string());
let msg = response
.message
.unwrap_or_else(|| "Unknown error".to_string());
anyhow::bail!("{}", msg)
}
"shutdown" => {
Ok(serde_json::json!({"status": "shutdown"}))
}
_ => {
Ok(response.data.unwrap_or(serde_json::json!({})))
}
"shutdown" => Ok(serde_json::json!({"status": "shutdown"})),
_ => Ok(response.data.unwrap_or(serde_json::json!({}))),
}
}
@@ -377,8 +365,8 @@ pub fn send_typed_command<T: for<'de> Deserialize<'de>>(
command: &str,
args: Option<serde_json::Value>,
) -> Result<BridgeResponse<T>> {
let mut stream = TcpStream::connect(format!("127.0.0.1:{}", port))
.context("Failed to connect to bridge")?;
let mut stream =
TcpStream::connect(format!("127.0.0.1:{}", port)).context("Failed to connect to bridge")?;
stream.set_read_timeout(Some(Duration::from_secs(300))).ok();
stream.set_write_timeout(Some(Duration::from_secs(30))).ok();
@@ -475,10 +463,7 @@ fn find_headless_script(ghidra_install_dir: &Path) -> Result<PathBuf> {
if script_path.exists() {
Ok(script_path)
} else {
anyhow::bail!(
"analyzeHeadless not found at: {}",
support_dir.display()
)
anyhow::bail!("analyzeHeadless not found at: {}", support_dir.display())
}
}
-174
View File
@@ -1,174 +0,0 @@
//! Data structures for Ghidra query results.
//!
//! These are used to parse JSON responses from Ghidra scripts.
#![allow(dead_code)]
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Function {
pub name: String,
pub address: String,
pub size: u64,
pub signature: Option<String>,
pub entry_point: String,
pub calling_convention: Option<String>,
#[serde(default)]
pub parameters: Vec<Parameter>,
#[serde(default)]
pub local_variables: Vec<LocalVariable>,
#[serde(default)]
pub calls: Vec<String>,
#[serde(default)]
pub called_by: Vec<String>,
pub decompiled: Option<String>,
pub comment: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Parameter {
pub name: String,
pub data_type: String,
pub ordinal: u32,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LocalVariable {
pub name: String,
pub data_type: String,
pub stack_offset: Option<i32>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StringData {
pub address: String,
pub value: String,
pub length: usize,
pub encoding: String,
#[serde(default)]
pub references: Vec<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Symbol {
pub name: String,
pub address: String,
pub symbol_type: String,
pub namespace: Option<String>,
pub source: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Import {
pub name: String,
pub address: String,
pub library: String,
pub ordinal: Option<u32>,
pub is_external: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Export {
pub name: String,
pub address: String,
pub ordinal: Option<u32>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct XRef {
pub from: String,
pub to: String,
pub ref_type: String,
pub from_function: Option<String>,
pub to_function: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MemoryBlock {
pub name: String,
pub start: String,
pub end: String,
pub size: u64,
pub permissions: String,
pub is_initialized: bool,
pub is_loaded: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Section {
pub name: String,
pub address: String,
pub size: u64,
pub virtual_address: String,
pub file_offset: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Comment {
pub address: String,
pub comment_type: String,
pub text: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DataType {
pub name: String,
pub category: String,
pub size: Option<u64>,
pub description: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Instruction {
pub address: String,
pub mnemonic: String,
pub operands: String,
pub bytes: String,
pub length: u32,
pub flow_type: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BasicBlock {
pub start: String,
pub end: String,
pub size: u64,
pub instruction_count: u32,
#[serde(default)]
pub successors: Vec<String>,
#[serde(default)]
pub predecessors: Vec<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ProgramInfo {
pub name: String,
pub executable_path: String,
pub executable_format: String,
pub compiler: Option<String>,
pub language: String,
pub creation_date: Option<String>,
pub image_base: String,
pub min_address: String,
pub max_address: String,
pub function_count: usize,
pub instruction_count: usize,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum GhidraData {
Function(Function),
String(StringData),
Symbol(Symbol),
Import(Import),
Export(Export),
XRef(XRef),
MemoryBlock(MemoryBlock),
Section(Section),
Comment(Comment),
DataType(DataType),
Instruction(Instruction),
BasicBlock(BasicBlock),
}
-4
View File
@@ -1,8 +1,6 @@
#![allow(dead_code)]
pub mod bridge;
pub mod data;
pub mod scripts;
pub mod setup;
use crate::config::Config;
@@ -42,13 +40,11 @@ impl GhidraClient {
#[cfg(target_os = "windows")]
{
// Use analyzeHeadless with Jython support
support_dir.join("analyzeHeadless.bat")
}
#[cfg(not(target_os = "windows"))]
{
// Use analyzeHeadless with Jython support
support_dir.join("analyzeHeadless")
}
}
-385
View File
@@ -1,385 +0,0 @@
//! Built-in Ghidra scripts for data extraction
//! These are Python scripts that will be written to disk and executed by Ghidra headless
pub fn get_list_functions_script() -> &'static str {
r#"
# List all functions in the program
# @category Analysis
# @runtime Jython
import json
functions = []
function_manager = currentProgram.getFunctionManager()
for func in function_manager.getFunctions(True):
entry = func.getEntryPoint()
body = func.getBody()
func_data = {
"name": func.getName(),
"address": entry.toString(),
"size": body.getNumAddresses(),
"entry_point": entry.toString(),
"signature": func.getPrototypeString(False, False) if func.getSignature() else None,
"calling_convention": func.getCallingConventionName(),
"comment": func.getComment()
}
# Get called functions
called = []
refs = func.getBody().getAddresses(True)
for addr in refs:
for ref in currentProgram.getReferenceManager().getReferencesFrom(addr):
if ref.getReferenceType().isCall():
to_addr = ref.getToAddress()
to_func = function_manager.getFunctionAt(to_addr)
if to_func:
called.append(to_func.getName())
func_data["calls"] = list(set(called))
# Get callers
callers = []
refs_to = currentProgram.getReferenceManager().getReferencesTo(entry)
for ref in refs_to:
if ref.getReferenceType().isCall():
from_addr = ref.getFromAddress()
from_func = function_manager.getFunctionContaining(from_addr)
if from_func:
callers.append(from_func.getName())
func_data["called_by"] = list(set(callers))
functions.append(func_data)
print("---GHIDRA_CLI_START---")
print(json.dumps(functions, indent=2))
print("---GHIDRA_CLI_END---")
"#
}
pub fn get_decompile_function_script() -> &'static str {
r#"
# Decompile a specific function by address or name
# @category Analysis
# @runtime Jython
import json
from ghidra.app.decompiler import DecompInterface
from ghidra.util.task import ConsoleTaskMonitor
# Get function target from args (can be address or name)
script_args = getScriptArgs()
if len(script_args) < 1:
print("---GHIDRA_CLI_START---")
print(json.dumps({"error": "No target provided. Specify an address (0x...) or function name."}))
print("---GHIDRA_CLI_END---")
exit(1)
target = script_args[0]
function_manager = currentProgram.getFunctionManager()
func = None
# Try to parse as address first
if target.startswith("0x") or target.startswith("0X"):
# It's an address
addr = currentProgram.getAddressFactory().getAddress(target)
if addr:
func = function_manager.getFunctionContaining(addr)
elif target.isdigit() or (len(target) > 1 and target[0].isdigit()):
# Might be a hex address without 0x prefix
try:
addr = currentProgram.getAddressFactory().getAddress(target)
if addr:
func = function_manager.getFunctionContaining(addr)
except:
pass
# If not found by address, try by name
if not func:
# Search for function by name (exact match first)
for f in function_manager.getFunctions(True):
if f.getName() == target:
func = f
break
# If still not found, try partial match
if not func:
for f in function_manager.getFunctions(True):
if target in f.getName():
func = f
break
if not func:
print("---GHIDRA_CLI_START---")
print(json.dumps({"error": "No function found for: " + target}))
print("---GHIDRA_CLI_END---")
exit(1)
# Decompile
decompiler = DecompInterface()
decompiler.openProgram(currentProgram)
monitor = ConsoleTaskMonitor()
results = decompiler.decompileFunction(func, 30, monitor)
if results.decompileCompleted():
code = results.getDecompiledFunction().getC()
result = {
"name": func.getName(),
"address": func.getEntryPoint().toString(),
"signature": func.getPrototypeString(False, False),
"code": code
}
print("---GHIDRA_CLI_START---")
print(json.dumps(result, indent=2))
print("---GHIDRA_CLI_END---")
else:
print("---GHIDRA_CLI_START---")
print(json.dumps({"error": "Decompilation failed for: " + func.getName()}))
print("---GHIDRA_CLI_END---")
"#
}
pub fn get_list_strings_script() -> &'static str {
r#"
# List all strings in the program
# @category Analysis
# @runtime Jython
import json
strings = []
listing = currentProgram.getListing()
data_iterator = listing.getDefinedData(True)
while data_iterator.hasNext():
data = data_iterator.next()
if data.hasStringValue():
try:
# Get string value, handle Unicode properly
string_val = unicode(data.getValue())
string_data = {
"address": str(data.getAddress()),
"value": string_val,
"length": len(string_val),
"encoding": "unicode"
}
# Get references to this string
refs = []
refs_to = currentProgram.getReferenceManager().getReferencesTo(data.getAddress())
for ref in refs_to:
refs.append(str(ref.getFromAddress()))
string_data["references"] = refs
strings.append(string_data)
except Exception as e:
# Skip strings that cause encoding issues
pass
print("---GHIDRA_CLI_START---")
print(json.dumps(strings, indent=2))
print("---GHIDRA_CLI_END---")
"#
}
pub fn get_list_imports_script() -> &'static str {
r#"
# List all imports in the program
# @category Analysis
# @runtime Jython
import json
imports = []
symbol_table = currentProgram.getSymbolTable()
external_manager = currentProgram.getExternalManager()
for symbol in symbol_table.getExternalSymbols():
external_location = external_manager.getExternalLocation(symbol)
if external_location:
import_data = {
"name": symbol.getName(),
"address": symbol.getAddress().toString(),
"library": external_location.getLibraryName(),
"is_external": True
}
imports.append(import_data)
print("---GHIDRA_CLI_START---")
print(json.dumps(imports, indent=2))
print("---GHIDRA_CLI_END---")
"#
}
pub fn get_list_exports_script() -> &'static str {
r#"
# List all exports in the program
# @category Analysis
# @runtime Jython
import json
exports = []
symbol_table = currentProgram.getSymbolTable()
for symbol in symbol_table.getSymbolIterator():
if symbol.isExternalEntryPoint():
export_data = {
"name": symbol.getName(),
"address": symbol.getAddress().toString()
}
exports.append(export_data)
print("---GHIDRA_CLI_START---")
print(json.dumps(exports, indent=2))
print("---GHIDRA_CLI_END---")
"#
}
pub fn get_memory_map_script() -> &'static str {
r#"
# Get memory map
# @category Analysis
# @runtime Jython
import json
blocks = []
memory = currentProgram.getMemory()
for block in memory.getBlocks():
block_data = {
"name": block.getName(),
"start": block.getStart().toString(),
"end": block.getEnd().toString(),
"size": block.getSize(),
"permissions": "",
"is_initialized": block.isInitialized(),
"is_loaded": block.isLoaded()
}
# Build permissions string
perms = ""
if block.isRead():
perms += "r"
if block.isWrite():
perms += "w"
if block.isExecute():
perms += "x"
block_data["permissions"] = perms
blocks.append(block_data)
print("---GHIDRA_CLI_START---")
print(json.dumps(blocks, indent=2))
print("---GHIDRA_CLI_END---")
"#
}
pub fn get_program_info_script() -> &'static str {
r#"
# Get program information
# @category Analysis
# @runtime Jython
import json
info = {
"name": currentProgram.getName(),
"executable_path": currentProgram.getExecutablePath(),
"executable_format": currentProgram.getExecutableFormat(),
"compiler": currentProgram.getCompiler() if currentProgram.getCompiler() else None,
"language": currentProgram.getLanguage().toString(),
"image_base": currentProgram.getImageBase().toString(),
"min_address": currentProgram.getMinAddress().toString(),
"max_address": currentProgram.getMaxAddress().toString()
}
# Count functions and instructions
function_manager = currentProgram.getFunctionManager()
info["function_count"] = function_manager.getFunctionCount()
instruction_count = 0
listing = currentProgram.getListing()
for instruction in listing.getInstructions(True):
instruction_count += 1
info["instruction_count"] = instruction_count
print("---GHIDRA_CLI_START---")
print(json.dumps(info, indent=2))
print("---GHIDRA_CLI_END---")
"#
}
pub fn get_xrefs_to_script() -> &'static str {
r#"
# Get cross-references to an address
# @category Analysis
# @runtime Jython
import json
script_args = getScriptArgs()
if len(script_args) < 1:
print("---GHIDRA_CLI_START---")
print(json.dumps({"error": "No address provided"}))
print("---GHIDRA_CLI_END---")
exit(1)
addr_str = script_args[0]
addr = currentProgram.getAddressFactory().getAddress(addr_str)
xrefs = []
refs = currentProgram.getReferenceManager().getReferencesTo(addr)
function_manager = currentProgram.getFunctionManager()
for ref in refs:
from_addr = ref.getFromAddress()
from_func = function_manager.getFunctionContaining(from_addr)
to_func = function_manager.getFunctionContaining(addr)
xref_data = {
"from": from_addr.toString(),
"to": addr.toString(),
"ref_type": ref.getReferenceType().toString(),
"from_function": from_func.getName() if from_func else None,
"to_function": to_func.getName() if to_func else None
}
xrefs.append(xref_data)
print("---GHIDRA_CLI_START---")
print(json.dumps(xrefs, indent=2))
print("---GHIDRA_CLI_END---")
"#
}
/// Save a script to disk
pub fn save_script(
name: &str,
content: &str,
scripts_dir: &std::path::Path,
) -> crate::error::Result<std::path::PathBuf> {
// All scripts are Python now with PyGhidra support
let script_path = scripts_dir.join(format!("{}.py", name));
std::fs::write(&script_path, content)?;
Ok(script_path)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_scripts_not_empty() {
assert!(!get_list_functions_script().is_empty());
assert!(!get_decompile_function_script().is_empty());
assert!(!get_list_strings_script().is_empty());
}
}
-22
View File
@@ -1,22 +0,0 @@
# Batch operations script
# @category CLI
#
# Note: Batch operations are handled directly in Rust handler.
# This script exists for consistency but is not actively used.
import sys
import json
def batch_placeholder():
"""Placeholder function - batch operations handled in Rust."""
return {"error": "Batch operations are handled by the Rust daemon, not via Python script"}
if __name__ == "__main__":
try:
print("---GHIDRA_CLI_START---")
print(json.dumps(batch_placeholder()))
print("---GHIDRA_CLI_END---")
except Exception as e:
print("---GHIDRA_CLI_START---")
print(json.dumps({"error": str(e)}))
print("---GHIDRA_CLI_END---")
File diff suppressed because it is too large Load Diff
-170
View File
@@ -1,170 +0,0 @@
# Comment operations script
# @category CLI
import sys
import json
def list_comments():
"""List all comments in the program."""
if currentProgram is None:
return {"error": "No program loaded"}
listing = currentProgram.getListing()
comments = []
# Iterate over all memory blocks to handle multiple address spaces
from ghidra.program.model.address import AddressSet
memory = currentProgram.getMemory()
for block in memory.getBlocks():
# Create an AddressSet for this block
address_set = AddressSet(block.getStart(), block.getEnd())
# Get comment addresses in this block
code_unit_iter = listing.getCommentAddressIterator(address_set, True)
for addr in code_unit_iter:
code_unit = listing.getCodeUnitAt(addr)
if code_unit is None:
continue
from ghidra.program.model.listing import CodeUnit
comment_types = [
("EOL", CodeUnit.EOL_COMMENT),
("PRE", CodeUnit.PRE_COMMENT),
("POST", CodeUnit.POST_COMMENT),
("PLATE", CodeUnit.PLATE_COMMENT)
]
for comment_name, comment_type in comment_types:
text = code_unit.getComment(comment_type)
if text:
comments.append({
"address": str(addr),
"type": comment_name,
"text": text
})
return {"comments": comments, "count": len(comments)}
def get_comments(address_str):
"""Get comments at a specific address."""
if currentProgram is None:
return {"error": "No program loaded"}
try:
addr = currentProgram.getAddressFactory().getAddress(address_str)
if addr is None:
return {"error": "Invalid address: " + address_str}
listing = currentProgram.getListing()
code_unit = listing.getCodeUnitAt(addr)
if code_unit is None:
return {"error": "No code unit at address: " + address_str}
from ghidra.program.model.listing import CodeUnit
comments = []
comment_types = [
("EOL", CodeUnit.EOL_COMMENT),
("PRE", CodeUnit.PRE_COMMENT),
("POST", CodeUnit.POST_COMMENT),
("PLATE", CodeUnit.PLATE_COMMENT)
]
for comment_name, comment_type in comment_types:
text = code_unit.getComment(comment_type)
if text:
comments.append({
"type": comment_name,
"text": text
})
return {"address": address_str, "comments": comments}
except Exception as e:
return {"error": "Failed to get comments: " + str(e)}
def set_comment(address_str, text, comment_type_str):
"""Set a comment at a specific address."""
if currentProgram is None:
return {"error": "No program loaded"}
try:
addr = currentProgram.getAddressFactory().getAddress(address_str)
if addr is None:
return {"error": "Invalid address: " + address_str}
listing = currentProgram.getListing()
from ghidra.program.model.listing import CodeUnit
valid_types = {"EOL", "PRE", "POST", "PLATE"}
if comment_type_str not in valid_types:
return {"error": "Invalid comment type: " + comment_type_str + ". Must be one of: EOL, PRE, POST, PLATE"}
comment_type = CodeUnit.EOL_COMMENT
if comment_type_str == "PRE":
comment_type = CodeUnit.PRE_COMMENT
elif comment_type_str == "POST":
comment_type = CodeUnit.POST_COMMENT
elif comment_type_str == "PLATE":
comment_type = CodeUnit.PLATE_COMMENT
listing.setComment(addr, comment_type, text)
return {"status": "set", "address": address_str}
except Exception as e:
return {"error": "Failed to set comment: " + str(e)}
def delete_comment(address_str):
"""Delete all comments at a specific address."""
if currentProgram is None:
return {"error": "No program loaded"}
try:
addr = currentProgram.getAddressFactory().getAddress(address_str)
if addr is None:
return {"error": "Invalid address: " + address_str}
listing = currentProgram.getListing()
from ghidra.program.model.listing import CodeUnit
listing.setComment(addr, CodeUnit.EOL_COMMENT, None)
listing.setComment(addr, CodeUnit.PRE_COMMENT, None)
listing.setComment(addr, CodeUnit.POST_COMMENT, None)
listing.setComment(addr, CodeUnit.PLATE_COMMENT, None)
return {"status": "deleted", "address": address_str}
except Exception as e:
return {"error": "Failed to delete comment: " + str(e)}
if __name__ == "__main__":
try:
if len(args) < 1:
print("---GHIDRA_CLI_START---")
print(json.dumps({"error": "No command specified"}))
print("---GHIDRA_CLI_END---")
sys.exit(1)
command = args[0]
if command == "list":
result = list_comments()
elif command == "get":
result = get_comments(args[1] if len(args) > 1 else None)
elif command == "set":
text = args[2] if len(args) > 2 else ""
comment_type = args[3] if len(args) > 3 else "EOL"
result = set_comment(args[1] if len(args) > 1 else None, text, comment_type)
elif command == "delete":
result = delete_comment(args[1] if len(args) > 1 else None)
else:
result = {"error": "Unknown command: " + command}
print("---GHIDRA_CLI_START---")
print(json.dumps(result))
print("---GHIDRA_CLI_END---")
except Exception as e:
print("---GHIDRA_CLI_START---")
print(json.dumps({"error": str(e)}))
print("---GHIDRA_CLI_END---")
-132
View File
@@ -1,132 +0,0 @@
# Diff operations script
# @category CLI
import sys
import json
def diff_programs(prog1, prog2):
"""Compare two programs structurally."""
if currentProgram is None:
return {"error": "No program loaded"}
try:
func_manager = currentProgram.getFunctionManager()
memory = currentProgram.getMemory()
symbol_table = currentProgram.getSymbolTable()
prog1_stats = {
"name": prog1,
"function_count": func_manager.getFunctionCount(),
"memory_size": memory.getSize(),
"symbol_count": symbol_table.getNumSymbols()
}
memory_blocks = []
for block in memory.getBlocks():
memory_blocks.append({
"name": block.getName(),
"start": str(block.getStart()),
"end": str(block.getEnd()),
"size": block.getSize()
})
prog1_stats["memory_blocks"] = memory_blocks
return {
"program1": prog1_stats,
"program2": {"name": prog2, "note": "Comparison requires loading second program"},
"status": "partial",
"message": "Single program stats returned (multi-program comparison not implemented)"
}
except Exception as e:
return {"error": "Failed to diff programs: " + str(e)}
def diff_functions(func1, func2):
"""Compare two functions by decompilation."""
if currentProgram is None:
return {"error": "No program loaded"}
try:
from ghidra.app.decompiler import DecompInterface
func_manager = currentProgram.getFunctionManager()
target_func1 = None
target_func2 = None
for func in func_manager.getFunctions(True):
if func.getName() == func1:
target_func1 = func
if func.getName() == func2:
target_func2 = func
if target_func1 is None:
return {"error": "Function not found: " + func1}
if target_func2 is None:
return {"error": "Function not found: " + func2}
decompiler = DecompInterface()
decompiler.openProgram(currentProgram)
result1 = decompiler.decompileFunction(target_func1, 30, monitor)
result2 = decompiler.decompileFunction(target_func2, 30, monitor)
if not result1.decompileCompleted():
return {"error": "Failed to decompile " + func1}
if not result2.decompileCompleted():
return {"error": "Failed to decompile " + func2}
code1 = result1.getDecompiledFunction().getC()
code2 = result2.getDecompiledFunction().getC()
lines1 = code1.split('\n')
lines2 = code2.split('\n')
diff_lines = []
max_lines = max(len(lines1), len(lines2))
for i in range(max_lines):
line1 = lines1[i] if i < len(lines1) else ""
line2 = lines2[i] if i < len(lines2) else ""
if line1 != line2:
diff_lines.append({
"line": i + 1,
"func1": line1,
"func2": line2,
"status": "changed"
})
return {
"func1": {"name": func1, "lines": len(lines1), "code": code1},
"func2": {"name": func2, "lines": len(lines2), "code": code2},
"differences": diff_lines,
"diff_count": len(diff_lines)
}
except Exception as e:
return {"error": "Failed to diff functions: " + str(e)}
if __name__ == "__main__":
try:
if len(args) < 1:
print("---GHIDRA_CLI_START---")
print(json.dumps({"error": "No command specified"}))
print("---GHIDRA_CLI_END---")
sys.exit(1)
command = args[0]
if command == "diff_programs":
result = diff_programs(args[1] if len(args) > 1 else "", args[2] if len(args) > 2 else "")
elif command == "diff_functions":
result = diff_functions(args[1] if len(args) > 1 else "", args[2] if len(args) > 2 else "")
else:
result = {"error": "Unknown command: " + command}
print("---GHIDRA_CLI_START---")
print(json.dumps(result))
print("---GHIDRA_CLI_END---")
except Exception as e:
print("---GHIDRA_CLI_START---")
print(json.dumps({"error": str(e)}))
print("---GHIDRA_CLI_END---")
-86
View File
@@ -1,86 +0,0 @@
# Disassembly script
# @category CLI
import sys
import json
def disassemble(address_str, count):
"""Disassemble instructions starting at address."""
if currentProgram is None:
return {"error": "No program loaded"}
try:
addr_factory = currentProgram.getAddressFactory()
if address_str.startswith("0x") or address_str.startswith("0X"):
address_str = address_str[2:]
addr = addr_factory.getAddress(address_str)
if addr is None:
return {"error": "Invalid address: " + address_str}
listing = currentProgram.getListing()
instruction = listing.getInstructionAt(addr)
if instruction is None:
return {"error": "No instruction at address: " + address_str}
results = []
current_instr = instruction
for i in range(count):
if current_instr is None:
break
instr_addr = current_instr.getAddress()
byte_array = current_instr.getBytes()
bytes_hex = ""
for b in byte_array:
bytes_hex += "{:02x}".format(b & 0xff)
mnemonic = current_instr.getMnemonicString()
operands = []
num_operands = current_instr.getNumOperands()
for j in range(num_operands):
operands.append(str(current_instr.getDefaultOperandRepresentation(j)))
results.append({
"address": str(instr_addr),
"bytes": bytes_hex,
"mnemonic": mnemonic,
"operands": operands
})
current_instr = current_instr.getNext()
return {"results": results, "count": len(results)}
except Exception as e:
return {"error": "Failed to disassemble: " + str(e)}
if __name__ == "__main__":
try:
if len(args) < 1:
print("---GHIDRA_CLI_START---")
print(json.dumps({"error": "No command specified"}))
print("---GHIDRA_CLI_END---")
sys.exit(1)
command = args[0]
if command == "disasm":
address = args[1] if len(args) > 1 else "0x0"
count = int(args[2]) if len(args) > 2 else 10
result = disassemble(address, count)
else:
result = {"error": "Unknown command: " + command}
print("---GHIDRA_CLI_START---")
print(json.dumps(result))
print("---GHIDRA_CLI_END---")
except Exception as e:
print("---GHIDRA_CLI_START---")
print(json.dumps({"error": str(e)}))
print("---GHIDRA_CLI_END---")
-259
View File
@@ -1,259 +0,0 @@
# Find/search operations script
# @category CLI
import sys
import json
def find_strings(pattern):
"""Find string references matching pattern."""
if currentProgram is None:
return {"error": "No program loaded"}
try:
listing = currentProgram.getListing()
results = []
data_iter = listing.getDefinedData(True)
while data_iter.hasNext():
data = data_iter.next()
if data.hasStringValue():
string_val = str(data.getValue())
if pattern.lower() in string_val.lower():
results.append({
"address": str(data.getAddress()),
"value": string_val,
"length": data.getLength()
})
return {"results": results, "count": len(results)}
except Exception as e:
return {"error": "Failed to find strings: " + str(e)}
def find_bytes(hex_pattern):
"""Find byte patterns in memory."""
if currentProgram is None:
return {"error": "No program loaded"}
try:
hex_clean = hex_pattern.replace("0x", "").replace(" ", "")
byte_array = []
for i in range(0, len(hex_clean), 2):
byte_val = int(hex_clean[i:i+2], 16)
if byte_val > 127:
byte_val = byte_val - 256
byte_array.append(byte_val)
from java.lang import Byte
search_bytes = [Byte(b) for b in byte_array]
memory = currentProgram.getMemory()
results = []
addr = memory.getMinAddress()
while addr is not None:
found_addr = memory.findBytes(addr, search_bytes, None, True, monitor)
if found_addr is None:
break
results.append({"address": str(found_addr)})
addr = found_addr.add(1)
if len(results) >= 100:
break
return {"results": results, "count": len(results)}
except Exception as e:
return {"error": "Failed to find bytes: " + str(e)}
def find_functions(pattern):
"""Find functions matching name pattern."""
if currentProgram is None:
return {"error": "No program loaded"}
try:
func_manager = currentProgram.getFunctionManager()
results = []
for func in func_manager.getFunctions(True):
func_name = func.getName()
if "*" in pattern:
import fnmatch
if fnmatch.fnmatch(func_name, pattern):
results.append({
"name": func_name,
"address": str(func.getEntryPoint()),
"size": func.getBody().getNumAddresses()
})
elif pattern.lower() in func_name.lower():
results.append({
"name": func_name,
"address": str(func.getEntryPoint()),
"size": func.getBody().getNumAddresses()
})
return {"results": results, "count": len(results)}
except Exception as e:
return {"error": "Failed to find functions: " + str(e)}
def find_calls(func_name):
"""Find all calls to a specific function."""
if currentProgram is None:
return {"error": "No program loaded"}
try:
func_manager = currentProgram.getFunctionManager()
target_func = None
for func in func_manager.getFunctions(True):
if func.getName() == func_name:
target_func = func
break
if target_func is None:
return {"error": "Function not found: " + func_name}
ref_manager = currentProgram.getReferenceManager()
target_addr = target_func.getEntryPoint()
refs = ref_manager.getReferencesTo(target_addr)
results = []
for ref in refs:
if ref.getReferenceType().isCall():
from_addr = ref.getFromAddress()
from_func = func_manager.getFunctionContaining(from_addr)
caller_name = "unknown"
if from_func is not None:
caller_name = from_func.getName()
results.append({
"address": str(from_addr),
"caller": caller_name,
"type": str(ref.getReferenceType())
})
return {"results": results, "count": len(results), "target": func_name}
except Exception as e:
return {"error": "Failed to find calls: " + str(e)}
def find_crypto():
"""Find potential crypto constants."""
if currentProgram is None:
return {"error": "No program loaded"}
try:
memory = currentProgram.getMemory()
results = []
crypto_patterns = {
"AES S-box": "637c777bf26b6fc53001672bfed7ab76",
"SHA-256": "428a2f98d728ae227137449123ef65cd",
"MD5": "d76aa478e8c7b756242070db01234567",
}
for name, pattern in crypto_patterns.items():
hex_clean = pattern.replace(" ", "")
byte_array = []
for i in range(0, len(hex_clean), 2):
byte_val = int(hex_clean[i:i+2], 16)
if byte_val > 127:
byte_val = byte_val - 256
byte_array.append(byte_val)
from java.lang import Byte
search_bytes = [Byte(b) for b in byte_array]
addr = memory.getMinAddress()
found_addr = memory.findBytes(addr, search_bytes, None, True, monitor)
if found_addr is not None:
results.append({
"type": name,
"address": str(found_addr),
"pattern": pattern
})
return {"results": results, "count": len(results)}
except Exception as e:
return {"error": "Failed to find crypto: " + str(e)}
def find_interesting():
"""Find interesting functions using heuristics."""
if currentProgram is None:
return {"error": "No program loaded"}
try:
func_manager = currentProgram.getFunctionManager()
ref_manager = currentProgram.getReferenceManager()
results = []
suspicious_names = ["password", "key", "encrypt", "decrypt", "crypt", "auth", "login", "admin", "secret"]
for func in func_manager.getFunctions(True):
func_name = func.getName()
func_addr = func.getEntryPoint()
func_size = func.getBody().getNumAddresses()
xref_count = len(list(ref_manager.getReferencesTo(func_addr)))
reasons = []
if func_size > 1000:
reasons.append("large function ({} bytes)".format(func_size))
if xref_count > 50:
reasons.append("many xrefs ({})".format(xref_count))
for sus_name in suspicious_names:
if sus_name in func_name.lower():
reasons.append("suspicious name")
break
if reasons:
results.append({
"name": func_name,
"address": str(func_addr),
"size": func_size,
"xrefs": xref_count,
"reasons": reasons
})
results.sort(key=lambda x: len(x["reasons"]), reverse=True)
return {"results": results[:50], "count": len(results)}
except Exception as e:
return {"error": "Failed to find interesting functions: " + str(e)}
if __name__ == "__main__":
try:
if len(args) < 1:
print("---GHIDRA_CLI_START---")
print(json.dumps({"error": "No command specified"}))
print("---GHIDRA_CLI_END---")
sys.exit(1)
command = args[0]
if command == "find_string":
result = find_strings(args[1] if len(args) > 1 else "")
elif command == "find_bytes":
result = find_bytes(args[1] if len(args) > 1 else "")
elif command == "find_function":
result = find_functions(args[1] if len(args) > 1 else "")
elif command == "find_calls":
result = find_calls(args[1] if len(args) > 1 else "")
elif command == "find_crypto":
result = find_crypto()
elif command == "find_interesting":
result = find_interesting()
else:
result = {"error": "Unknown command: " + command}
print("---GHIDRA_CLI_START---")
print(json.dumps(result))
print("---GHIDRA_CLI_END---")
except Exception as e:
print("---GHIDRA_CLI_START---")
print(json.dumps({"error": str(e)}))
print("---GHIDRA_CLI_END---")
-223
View File
@@ -1,223 +0,0 @@
# Graph operations script
# @category CLI
import sys
import json
def get_call_graph(limit):
"""Build full call graph."""
if currentProgram is None:
return {"error": "No program loaded"}
function_manager = currentProgram.getFunctionManager()
reference_manager = currentProgram.getReferenceManager()
nodes = []
edges = []
count = 0
for func in function_manager.getFunctions(True):
if limit and count >= limit:
break
func_addr = str(func.getEntryPoint())
nodes.append({
"id": func_addr,
"name": func.getName(),
"address": func_addr
})
from ghidra.program.model.symbol import RefType
refs = reference_manager.getReferencesFrom(func.getEntryPoint())
for ref in refs:
if ref.getReferenceType().isCall():
target_addr = ref.getToAddress()
target_func = function_manager.getFunctionAt(target_addr)
if target_func:
edges.append({
"from": func_addr,
"to": str(target_addr),
"type": "call"
})
count += 1
return {"nodes": nodes, "edges": edges, "node_count": len(nodes), "edge_count": len(edges)}
def get_callers(function_name, depth):
"""Get functions that call the specified function."""
if currentProgram is None:
return {"error": "No program loaded"}
function_manager = currentProgram.getFunctionManager()
reference_manager = currentProgram.getReferenceManager()
target_func = None
if function_name.startswith("0x") or all(c in "0123456789abcdefABCDEF" for c in function_name):
addr = currentProgram.getAddressFactory().getAddress(function_name)
if addr:
target_func = function_manager.getFunctionAt(addr)
else:
for func in function_manager.getFunctions(True):
if func.getName() == function_name:
target_func = func
break
if not target_func:
return {"error": "Function not found: " + function_name}
callers = []
visited = set()
def find_callers(func, current_depth):
if depth and current_depth >= depth:
return
if str(func.getEntryPoint()) in visited:
return
visited.add(str(func.getEntryPoint()))
from ghidra.program.model.symbol import RefType
refs = reference_manager.getReferencesTo(func.getEntryPoint())
for ref in refs:
if ref.getReferenceType().isCall():
from_addr = ref.getFromAddress()
caller_func = function_manager.getFunctionContaining(from_addr)
if caller_func:
caller_info = {
"name": caller_func.getName(),
"address": str(caller_func.getEntryPoint()),
"call_site": str(from_addr),
"depth": current_depth
}
callers.append(caller_info)
if depth is None or current_depth + 1 < depth:
find_callers(caller_func, current_depth + 1)
find_callers(target_func, 0)
return {"function": function_name, "callers": callers, "count": len(callers)}
def get_callees(function_name, depth):
"""Get functions called by the specified function."""
if currentProgram is None:
return {"error": "No program loaded"}
function_manager = currentProgram.getFunctionManager()
reference_manager = currentProgram.getReferenceManager()
target_func = None
if function_name.startswith("0x") or all(c in "0123456789abcdefABCDEF" for c in function_name):
addr = currentProgram.getAddressFactory().getAddress(function_name)
if addr:
target_func = function_manager.getFunctionAt(addr)
else:
for func in function_manager.getFunctions(True):
if func.getName() == function_name:
target_func = func
break
if not target_func:
return {"error": "Function not found: " + function_name}
callees = []
visited = set()
def find_callees(func, current_depth):
if depth and current_depth >= depth:
return
if str(func.getEntryPoint()) in visited:
return
visited.add(str(func.getEntryPoint()))
from ghidra.program.model.symbol import RefType
refs = reference_manager.getReferencesFrom(func.getEntryPoint())
for ref in refs:
if ref.getReferenceType().isCall():
to_addr = ref.getToAddress()
callee_func = function_manager.getFunctionAt(to_addr)
if callee_func:
callee_info = {
"name": callee_func.getName(),
"address": str(callee_func.getEntryPoint()),
"call_site": str(ref.getFromAddress()),
"depth": current_depth
}
callees.append(callee_info)
if depth is None or current_depth + 1 < depth:
find_callees(callee_func, current_depth + 1)
find_callees(target_func, 0)
return {"function": function_name, "callees": callees, "count": len(callees)}
def export_graph(export_format):
"""Export call graph in specified format."""
if currentProgram is None:
return {"error": "No program loaded"}
graph_data = get_call_graph(None)
if "error" in graph_data:
return graph_data
if export_format == "json":
return graph_data
elif export_format == "dot":
lines = ["digraph CallGraph {"]
lines.append(' rankdir=LR;')
lines.append(' node [shape=box];')
for node in graph_data["nodes"]:
node_id = node["id"].replace(":", "_")
label = node["name"]
lines.append(' "{}" [label="{}"];'.format(node_id, label))
for edge in graph_data["edges"]:
from_id = edge["from"].replace(":", "_")
to_id = edge["to"].replace(":", "_")
lines.append(' "{}" -> "{}";'.format(from_id, to_id))
lines.append("}")
return {"format": "dot", "output": "\n".join(lines)}
else:
return {"error": "Unsupported format: " + export_format}
if __name__ == "__main__":
try:
if len(args) < 1:
print("---GHIDRA_CLI_START---")
print(json.dumps({"error": "No command specified"}))
print("---GHIDRA_CLI_END---")
sys.exit(1)
command = args[0]
if command == "calls":
limit = int(args[1]) if len(args) > 1 and args[1] else None
result = get_call_graph(limit)
elif command == "callers":
func_name = args[1] if len(args) > 1 else None
depth = int(args[2]) if len(args) > 2 and args[2] else None
result = get_callers(func_name, depth)
elif command == "callees":
func_name = args[1] if len(args) > 1 else None
depth = int(args[2]) if len(args) > 2 and args[2] else None
result = get_callees(func_name, depth)
elif command == "export":
fmt = args[1] if len(args) > 1 else "json"
result = export_graph(fmt)
else:
result = {"error": "Unknown command: " + command}
print("---GHIDRA_CLI_START---")
print(json.dumps(result))
print("---GHIDRA_CLI_END---")
except Exception as e:
print("---GHIDRA_CLI_START---")
print(json.dumps({"error": str(e)}))
print("---GHIDRA_CLI_END---")
-134
View File
@@ -1,134 +0,0 @@
# Patch operations script
# @category CLI
import sys
import json
def patch_bytes(address_str, hex_data):
"""Patch bytes at the specified address."""
if currentProgram is None:
return {"error": "No program loaded"}
try:
addr = currentProgram.getAddressFactory().getAddress(address_str)
if addr is None:
return {"error": "Invalid address: " + address_str}
hex_clean = hex_data.replace("0x", "").replace(" ", "")
byte_array = []
for i in range(0, len(hex_clean), 2):
byte_val = int(hex_clean[i:i+2], 16)
if byte_val > 127:
byte_val = byte_val - 256
byte_array.append(byte_val)
from java.lang import Byte
patch_bytes = [Byte(b) for b in byte_array]
memory = currentProgram.getMemory()
memory.setBytes(addr, patch_bytes)
return {
"status": "patched",
"address": str(addr),
"bytes": len(patch_bytes)
}
except Exception as e:
return {"error": "Failed to patch bytes: " + str(e)}
def patch_nop(address_str):
"""NOP out instruction at the specified address."""
if currentProgram is None:
return {"error": "No program loaded"}
try:
addr = currentProgram.getAddressFactory().getAddress(address_str)
if addr is None:
return {"error": "Invalid address: " + address_str}
listing = currentProgram.getListing()
instruction = listing.getInstructionAt(addr)
if instruction is None:
return {"error": "No instruction at address: " + address_str}
instr_length = instruction.getLength()
processor = currentProgram.getLanguage().getProcessor().toString()
if "x86" in processor.lower():
nop_byte = 0x90
elif "ARM" in processor or "aarch" in processor.lower():
nop_byte = 0x00
else:
nop_byte = 0x00
from java.lang import Byte
nop_bytes = [Byte(nop_byte) for _ in range(instr_length)]
memory = currentProgram.getMemory()
memory.setBytes(addr, nop_bytes)
return {
"status": "nopped",
"address": str(addr),
"bytes": instr_length
}
except Exception as e:
return {"error": "Failed to NOP instruction: " + str(e)}
def export_binary(output_path):
"""Export the patched binary."""
if currentProgram is None:
return {"error": "No program loaded"}
try:
from ghidra.app.util.exporter import BinaryExporter
from java.io import File
exporter = BinaryExporter()
output_file = File(output_path)
exporter.export(output_file, currentProgram, None, monitor)
return {
"status": "exported",
"output": output_path
}
except Exception as e:
return {"error": "Failed to export binary: " + str(e)}
# Alias for bridge.py compatibility
def export_patches(output_path):
"""Export patches (alias for export_binary)."""
return export_binary(output_path)
if __name__ == "__main__":
try:
args = getScriptArgs()
if len(args) < 1:
print("---GHIDRA_CLI_START---")
print(json.dumps({"error": "No command specified"}))
print("---GHIDRA_CLI_END---")
sys.exit(1)
command = args[0]
if command == "patch_bytes":
result = patch_bytes(args[1] if len(args) > 1 else "", args[2] if len(args) > 2 else "")
elif command == "patch_nop":
result = patch_nop(args[1] if len(args) > 1 else "")
elif command == "patch_export":
result = export_binary(args[1] if len(args) > 1 else "")
else:
result = {"error": "Unknown command: " + command}
print("---GHIDRA_CLI_START---")
print(json.dumps(result))
print("---GHIDRA_CLI_END---")
except Exception as e:
print("---GHIDRA_CLI_START---")
print(json.dumps({"error": str(e)}))
print("---GHIDRA_CLI_END---")
-115
View File
@@ -1,115 +0,0 @@
# Program operations script
# @category CLI
import sys
import json
def close_program():
"""Close the current program."""
if currentProgram is None:
return {"error": "No program loaded"}
program_name = currentProgram.getName()
state.getTool().closeProgram(currentProgram, False)
return {"status": "closed", "program": program_name}
def delete_program(program_name):
"""Delete a program from the project."""
project = state.getProject()
if project is None:
return {"error": "No project open"}
project_data = project.getProjectData()
try:
program_file = project_data.getFile(program_name)
if program_file is None:
return {"error": "Program not found: " + program_name}
project_data.deleteFile(program_name)
return {"status": "deleted", "program": program_name}
except Exception as e:
return {"error": "Failed to delete program: " + str(e)}
def get_program_info():
"""Get current program metadata."""
if currentProgram is None:
return {"error": "No program loaded"}
info = {
"name": currentProgram.getName(),
"path": currentProgram.getExecutablePath(),
"format": currentProgram.getExecutableFormat(),
"processor": str(currentProgram.getLanguage().getProcessor()),
"language": str(currentProgram.getLanguage()),
"compiler": currentProgram.getCompiler() if currentProgram.getCompiler() else None,
"image_base": str(currentProgram.getImageBase()),
"min_address": str(currentProgram.getMinAddress()),
"max_address": str(currentProgram.getMaxAddress()),
"creation_date": str(currentProgram.getCreationDate())
}
return info
def export_program(export_format, output_path):
"""Export program to specified format."""
if currentProgram is None:
return {"error": "No program loaded"}
from ghidra.app.util.exporter import Exporter
from ghidra.framework.model import DomainFile
from java.io import File
if export_format == "json":
data = get_program_info()
function_manager = currentProgram.getFunctionManager()
functions = []
for func in function_manager.getFunctions(True):
functions.append({
"name": func.getName(),
"address": str(func.getEntryPoint()),
"size": func.getBody().getNumAddresses()
})
data["functions"] = functions
if output_path:
with open(output_path, 'w') as f:
json.dump(data, f, indent=2)
return {"status": "exported", "format": "json", "output": output_path}
else:
return data
else:
return {"error": "Unsupported export format: " + export_format}
if __name__ == "__main__":
try:
if len(args) < 1:
print("---GHIDRA_CLI_START---")
print(json.dumps({"error": "No command specified"}))
print("---GHIDRA_CLI_END---")
sys.exit(1)
command = args[0]
if command == "close":
result = close_program()
elif command == "delete":
result = delete_program(args[1] if len(args) > 1 else None)
elif command == "info":
result = get_program_info()
elif command == "export":
fmt = args[1] if len(args) > 1 else "json"
output = args[2] if len(args) > 2 else None
result = export_program(fmt, output)
else:
result = {"error": "Unknown command: " + command}
print("---GHIDRA_CLI_START---")
print(json.dumps(result))
print("---GHIDRA_CLI_END---")
except Exception as e:
print("---GHIDRA_CLI_START---")
print(json.dumps({"error": str(e)}))
print("---GHIDRA_CLI_END---")
-121
View File
@@ -1,121 +0,0 @@
# Script execution operations
# @category CLI
import sys
import json
import os
def run_script(script_path, script_args):
"""Run a user script file."""
if currentProgram is None:
return {"error": "No program loaded"}
try:
if not os.path.exists(script_path):
return {"error": "Script not found: " + script_path}
from ghidra.app.script import GhidraScriptUtil
script_info = GhidraScriptUtil.findScriptByName(os.path.basename(script_path))
if script_info is None:
return {"error": "Could not load script: " + script_path}
result = runScript(script_path, script_args if script_args else [])
return {
"status": "executed",
"script": script_path,
"result": str(result) if result is not None else None
}
except Exception as e:
return {"error": "Failed to run script: " + str(e)}
def exec_python(code):
"""Execute inline Python code."""
if currentProgram is None:
return {"error": "No program loaded"}
try:
local_vars = {
"currentProgram": currentProgram,
"currentAddress": currentAddress if 'currentAddress' in dir() else None,
"currentLocation": currentLocation if 'currentLocation' in dir() else None,
"state": state if 'state' in dir() else None
}
exec(code, globals(), local_vars)
output = local_vars.get("output", None)
return {
"status": "executed",
"output": str(output) if output is not None else "Code executed successfully"
}
except Exception as e:
return {"error": "Failed to execute Python code: " + str(e)}
def exec_java(code):
"""Execute inline Java code."""
if currentProgram is None:
return {"error": "No program loaded"}
try:
return {"error": "Java execution not yet implemented"}
except Exception as e:
return {"error": "Failed to execute Java code: " + str(e)}
def list_scripts():
"""List available scripts."""
try:
from ghidra.app.script import GhidraScriptUtil
script_infos = GhidraScriptUtil.getScriptSourceDirectories()
scripts = []
for script_dir in script_infos:
script_path = str(script_dir)
if os.path.exists(script_path) and os.path.isdir(script_path):
for filename in os.listdir(script_path):
if filename.endswith('.py') or filename.endswith('.java'):
scripts.append({
"name": filename,
"path": os.path.join(script_path, filename),
"type": "python" if filename.endswith('.py') else "java"
})
return {"scripts": scripts, "count": len(scripts)}
except Exception as e:
return {"error": "Failed to list scripts: " + str(e)}
if __name__ == "__main__":
try:
if len(args) < 1:
print("---GHIDRA_CLI_START---")
print(json.dumps({"error": "No command specified"}))
print("---GHIDRA_CLI_END---")
sys.exit(1)
command = args[0]
if command == "run":
script_path = args[1] if len(args) > 1 else None
script_args = args[2:] if len(args) > 2 else []
result = run_script(script_path, script_args)
elif command == "python":
code = args[1] if len(args) > 1 else None
result = exec_python(code)
elif command == "java":
code = args[1] if len(args) > 1 else None
result = exec_java(code)
elif command == "list":
result = list_scripts()
else:
result = {"error": "Unknown command: " + command}
print("---GHIDRA_CLI_START---")
print(json.dumps(result))
print("---GHIDRA_CLI_END---")
except Exception as e:
print("---GHIDRA_CLI_START---")
print(json.dumps({"error": str(e)}))
print("---GHIDRA_CLI_END---")
-98
View File
@@ -1,98 +0,0 @@
# Program statistics script
# @category CLI
import sys
import json
def get_stats():
"""Gather comprehensive program statistics."""
if currentProgram is None:
return {"error": "No program loaded"}
try:
func_manager = currentProgram.getFunctionManager()
symbol_table = currentProgram.getSymbolTable()
memory = currentProgram.getMemory()
data_type_manager = currentProgram.getDataTypeManager()
listing = currentProgram.getListing()
function_count = func_manager.getFunctionCount()
symbol_count = 0
symbol_iter = symbol_table.getAllSymbols(True)
while symbol_iter.hasNext():
symbol_iter.next()
symbol_count += 1
string_count = 0
data_iter = listing.getDefinedData(True)
while data_iter.hasNext():
data = data_iter.next()
if data.hasStringValue():
string_count += 1
memory_size = 0
for block in memory.getBlocks():
memory_size += block.getSize()
section_count = len(list(memory.getBlocks()))
import_count = 0
export_count = 0
for symbol in symbol_table.getExternalSymbols():
import_count += 1
export_iter = symbol_table.getExternalEntryPointIterator()
while export_iter.hasNext():
export_iter.next()
export_count += 1
data_type_count = data_type_manager.getDataTypeCount(False)
instruction_count = 0
code_unit_iter = listing.getInstructions(True)
while code_unit_iter.hasNext():
code_unit_iter.next()
instruction_count += 1
stats = {
"functions": function_count,
"symbols": symbol_count,
"strings": string_count,
"imports": import_count,
"exports": export_count,
"memory_size": memory_size,
"sections": section_count,
"data_types": data_type_count,
"instructions": instruction_count,
"program_name": currentProgram.getName(),
"executable_format": currentProgram.getExecutableFormat(),
"compiler": str(currentProgram.getCompiler()) if currentProgram.getCompiler() else "Unknown"
}
return {"stats": stats}
except Exception as e:
return {"error": "Failed to gather statistics: " + str(e)}
if __name__ == "__main__":
try:
if len(args) < 1:
print("---GHIDRA_CLI_START---")
print(json.dumps({"error": "No command specified"}))
print("---GHIDRA_CLI_END---")
sys.exit(1)
command = args[0]
if command == "stats":
result = get_stats()
else:
result = {"error": "Unknown command: " + command}
print("---GHIDRA_CLI_START---")
print(json.dumps(result))
print("---GHIDRA_CLI_END---")
except Exception as e:
print("---GHIDRA_CLI_START---")
print(json.dumps({"error": str(e)}))
print("---GHIDRA_CLI_END---")
-162
View File
@@ -1,162 +0,0 @@
# Symbol operations script
# @category CLI
import sys
import json
def list_symbols(name_filter):
"""List all symbols in the program."""
if currentProgram is None:
return {"error": "No program loaded"}
symbol_table = currentProgram.getSymbolTable()
symbols = []
for symbol in symbol_table.getAllSymbols(True):
name = symbol.getName()
if name_filter and name_filter.lower() not in name.lower():
continue
symbol_data = {
"name": name,
"address": str(symbol.getAddress()),
"type": str(symbol.getSymbolType()),
"source": str(symbol.getSource()),
"is_primary": symbol.isPrimary()
}
symbols.append(symbol_data)
return {"symbols": symbols, "count": len(symbols)}
def get_symbol(address_or_name):
"""Get symbol at specific address or by name."""
if currentProgram is None:
return {"error": "No program loaded"}
symbol_table = currentProgram.getSymbolTable()
if address_or_name.startswith("0x") or all(c in "0123456789abcdefABCDEF" for c in address_or_name):
try:
addr = currentProgram.getAddressFactory().getAddress(address_or_name)
if addr is None:
return {"error": "Invalid address: " + address_or_name}
symbols_at_addr = symbol_table.getSymbols(addr)
if not symbols_at_addr:
return {"error": "No symbol at address: " + address_or_name}
result_symbols = []
for symbol in symbols_at_addr:
result_symbols.append({
"name": symbol.getName(),
"address": str(symbol.getAddress()),
"type": str(symbol.getSymbolType()),
"source": str(symbol.getSource())
})
return {"symbols": result_symbols}
except Exception as e:
return {"error": "Failed to get symbol: " + str(e)}
else:
symbols = list(symbol_table.getSymbols(address_or_name))
if not symbols:
return {"error": "Symbol not found: " + address_or_name}
result_symbols = []
for symbol in symbols:
result_symbols.append({
"name": symbol.getName(),
"address": str(symbol.getAddress()),
"type": str(symbol.getSymbolType()),
"source": str(symbol.getSource())
})
return {"symbols": result_symbols}
def create_symbol(address_str, name):
"""Create a new symbol."""
if currentProgram is None:
return {"error": "No program loaded"}
try:
addr = currentProgram.getAddressFactory().getAddress(address_str)
if addr is None:
return {"error": "Invalid address: " + address_str}
symbol_table = currentProgram.getSymbolTable()
from ghidra.program.model.symbol import SourceType
symbol_table.createLabel(addr, name, SourceType.USER_DEFINED)
return {"status": "created", "address": address_str, "name": name}
except Exception as e:
return {"error": "Failed to create symbol: " + str(e)}
def delete_symbol(name):
"""Delete a symbol by name."""
if currentProgram is None:
return {"error": "No program loaded"}
try:
symbol_table = currentProgram.getSymbolTable()
symbols = list(symbol_table.getSymbols(name))
if not symbols:
return {"error": "Symbol not found: " + name}
for symbol in symbols:
symbol.delete()
return {"status": "deleted", "name": name}
except Exception as e:
return {"error": "Failed to delete symbol: " + str(e)}
def rename_symbol(old_name, new_name):
"""Rename a symbol."""
if currentProgram is None:
return {"error": "No program loaded"}
try:
symbol_table = currentProgram.getSymbolTable()
symbols = list(symbol_table.getSymbols(old_name))
if not symbols:
return {"error": "Symbol not found: " + old_name}
from ghidra.program.model.symbol import SourceType
for symbol in symbols:
symbol.setName(new_name, SourceType.USER_DEFINED)
return {"status": "renamed", "old_name": old_name, "new_name": new_name}
except Exception as e:
return {"error": "Failed to rename symbol: " + str(e)}
if __name__ == "__main__":
try:
if len(args) < 1:
print("---GHIDRA_CLI_START---")
print(json.dumps({"error": "No command specified"}))
print("---GHIDRA_CLI_END---")
sys.exit(1)
command = args[0]
if command == "list":
result = list_symbols(args[1] if len(args) > 1 else None)
elif command == "get":
result = get_symbol(args[1] if len(args) > 1 else None)
elif command == "create":
result = create_symbol(args[1] if len(args) > 1 else None, args[2] if len(args) > 2 else None)
elif command == "delete":
result = delete_symbol(args[1] if len(args) > 1 else None)
elif command == "rename":
result = rename_symbol(args[1] if len(args) > 1 else None, args[2] if len(args) > 2 else None)
else:
result = {"error": "Unknown command: " + command}
print("---GHIDRA_CLI_START---")
print(json.dumps(result))
print("---GHIDRA_CLI_END---")
except Exception as e:
print("---GHIDRA_CLI_START---")
print(json.dumps({"error": str(e)}))
print("---GHIDRA_CLI_END---")
-137
View File
@@ -1,137 +0,0 @@
# Type operations script
# @category CLI
import sys
import json
def list_types():
"""List all defined types in the program."""
if currentProgram is None:
return {"error": "No program loaded"}
data_type_manager = currentProgram.getDataTypeManager()
types = []
for data_type in data_type_manager.getAllDataTypes():
type_data = {
"name": data_type.getName(),
"path": data_type.getPathName(),
"category": data_type.getCategoryPath().toString(),
"size": data_type.getLength()
}
types.append(type_data)
return {"types": types, "count": len(types)}
def get_type(type_name):
"""Get type definition by name."""
if currentProgram is None:
return {"error": "No program loaded"}
data_type_manager = currentProgram.getDataTypeManager()
data_type = data_type_manager.getDataType(type_name)
if data_type is None:
for dt in data_type_manager.getAllDataTypes():
if dt.getName() == type_name:
data_type = dt
break
if data_type is None:
return {"error": "Type not found: " + type_name}
type_info = {
"name": data_type.getName(),
"path": data_type.getPathName(),
"category": data_type.getCategoryPath().toString(),
"size": data_type.getLength(),
"description": data_type.getDescription()
}
from ghidra.program.model.data import Structure, Union
if isinstance(data_type, Structure) or isinstance(data_type, Union):
components = []
for component in data_type.getComponents():
components.append({
"name": component.getFieldName(),
"type": component.getDataType().getName(),
"offset": component.getOffset(),
"size": component.getLength()
})
type_info["components"] = components
return type_info
def create_type(type_name):
"""Create a new empty struct type."""
if currentProgram is None:
return {"error": "No program loaded"}
try:
from ghidra.program.model.data import StructureDataType
data_type_manager = currentProgram.getDataTypeManager()
new_struct = StructureDataType(type_name, 0)
data_type_manager.addDataType(new_struct, None)
return {"status": "created", "name": type_name}
except Exception as e:
return {"error": "Failed to create type: " + str(e)}
def apply_type(address_str, type_name):
"""Apply a type to a specific address."""
if currentProgram is None:
return {"error": "No program loaded"}
try:
addr = currentProgram.getAddressFactory().getAddress(address_str)
if addr is None:
return {"error": "Invalid address: " + address_str}
data_type_manager = currentProgram.getDataTypeManager()
data_type = data_type_manager.getDataType(type_name)
if data_type is None:
for dt in data_type_manager.getAllDataTypes():
if dt.getName() == type_name:
data_type = dt
break
if data_type is None:
return {"error": "Type not found: " + type_name}
listing = currentProgram.getListing()
listing.createData(addr, data_type)
return {"status": "applied", "address": address_str, "type": type_name}
except Exception as e:
return {"error": "Failed to apply type: " + str(e)}
if __name__ == "__main__":
try:
if len(args) < 1:
print("---GHIDRA_CLI_START---")
print(json.dumps({"error": "No command specified"}))
print("---GHIDRA_CLI_END---")
sys.exit(1)
command = args[0]
if command == "list":
result = list_types()
elif command == "get":
result = get_type(args[1] if len(args) > 1 else None)
elif command == "create":
result = create_type(args[1] if len(args) > 1 else None)
elif command == "apply":
result = apply_type(args[1] if len(args) > 1 else None, args[2] if len(args) > 2 else None)
else:
result = {"error": "Unknown command: " + command}
print("---GHIDRA_CLI_START---")
print(json.dumps(result))
print("---GHIDRA_CLI_END---")
except Exception as e:
print("---GHIDRA_CLI_START---")
print(json.dumps({"error": str(e)}))
print("---GHIDRA_CLI_END---")

Some files were not shown because too many files have changed in this diff Show More