use crate::analysis::engine::Analysis;
use crate::analysis::thunks::bare_name;
use crate::model::{Arch, Binary, Format};
use iced_x86::{
Decoder, DecoderOptions, Instruction, InstructionInfoFactory, Mnemonic, OpAccess, OpKind,
Register,
};
use serde::Serialize;
use std::collections::{BTreeSet, VecDeque};
#[derive(Clone, Copy)]
enum Check {
Format(u8),
AllocSize(u8),
CopySize(u8),
Unbounded,
}
fn checks(api: &str) -> &'static [Check] {
use Check::*;
match api {
"printf" | "vprintf" => &[Format(1)],
"fprintf" | "vfprintf" | "syslog" | "wsprintfA" | "wsprintfW" => &[Format(2)],
"sprintf" | "vsprintf" => &[Format(2), Unbounded],
"snprintf" | "_snprintf" => &[Format(3)],
"malloc" => &[AllocSize(1)],
"realloc" | "VirtualAlloc" => &[AllocSize(2)],
"HeapAlloc" => &[AllocSize(3)],
"memcpy" | "memmove" | "memset" | "RtlCopyMemory" | "CopyMemory" | "bcopy" | "strncpy"
| "strncat" | "wcsncpy" => &[CopySize(3)],
"strcpy" | "strcat" | "gets" | "wcscpy" | "wcscat" | "lstrcpyA" | "lstrcpyW"
| "lstrcatA" | "lstrcatW" | "StrCpyA" | "StrCpyW" => &[Unbounded],
_ => &[],
}
}
#[derive(Debug, Clone, Serialize)]
pub struct Finding {
pub addr: u64,
pub func: Option<String>,
pub api: String,
pub pattern: &'static str,
pub severity: u8,
pub detail: String,
pub reachable: bool,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Origin {
Fixed,
Dynamic,
Multiply,
Subtract,
Constant,
Stack,
Unknown,
}
pub fn run(an: &Analysis, bin: &Binary, bytes: &[u8]) -> Vec<Finding> {
if !matches!(an.arch, Arch::X86 | Arch::X86_64) {
return Vec::new();
}
let win64 = bin.format == Format::Pe && an.bits == 64;
let reachable = reachable_functions(an, bin);
let mut out = Vec::new();
for f in &an.functions {
let from_entry = reachable.contains(&f.addr);
for block in &f.blocks {
for (i, ins) in block.insns.iter().enumerate() {
let Some(target) = ins.target else { continue };
let Some(full) = an.imports.get(&target).or_else(|| an.names.get(&target)) else {
continue;
};
let api = bare_name(full);
let cs = checks(api);
if cs.is_empty() {
continue;
}
for check in cs {
if let Some(mut fnd) =
classify(*check, api, an, bin, bytes, block, i, win64, from_entry)
{
fnd.func = Some(f.name.clone());
out.push(fnd);
}
}
}
}
}
out.sort_by(|a, b| {
b.severity
.cmp(&a.severity)
.then(b.reachable.cmp(&a.reachable))
.then(a.addr.cmp(&b.addr))
});
out
}
#[allow(clippy::too_many_arguments)]
fn classify(
check: Check,
api: &str,
an: &Analysis,
bin: &Binary,
bytes: &[u8],
block: &crate::analysis::engine::BasicBlock,
call_idx: usize,
win64: bool,
reachable: bool,
) -> Option<Finding> {
let addr = block.insns[call_idx].addr + an.display_base;
let mk = |pattern, severity, detail: String| Finding {
addr,
func: None,
api: api.to_string(),
pattern,
severity,
detail,
reachable,
};
let dest_is_stack =
provenance(an, bin, bytes, block, call_idx, 1, win64).origin == Origin::Stack;
match check {
Check::Unbounded => {
if dest_is_stack {
return Some(mk(
"stack-overflow",
3,
"unbounded copy into a stack buffer (classic stack overflow)".to_string(),
));
}
let sev = if reachable { 3 } else { 2 };
let detail = if reachable {
"unbounded copy reachable from an entry point or export".to_string()
} else {
"unbounded copy; no length argument bounds the write".to_string()
};
Some(mk("unbounded-copy", sev, detail))
}
Check::Format(arg) => {
match provenance(an, bin, bytes, block, call_idx, arg, win64).origin {
Origin::Dynamic => Some(mk(
"format-string",
3,
"format argument is loaded at runtime, not a constant string".to_string(),
)),
_ => None,
}
}
Check::AllocSize(arg) => {
let p = provenance(an, bin, bytes, block, call_idx, arg, win64);
match p.origin {
Origin::Multiply => Some(sized(
&mk,
"alloc-overflow",
3,
"allocation size computed by multiplication (integer overflow?)",
p.bounded,
)),
Origin::Subtract => Some(sized(
&mk,
"alloc-underflow",
2,
"allocation size computed by subtraction",
p.bounded,
)),
_ => None,
}
}
Check::CopySize(arg) => {
let p = provenance(an, bin, bytes, block, call_idx, arg, win64);
let dst = if dest_is_stack {
" into a stack buffer"
} else {
""
};
match p.origin {
Origin::Subtract => Some(sized(
&mk,
"copy-underflow",
3,
&format!(
"copy length computed by subtraction (integer underflow to a huge size?){dst}"
),
p.bounded,
)),
Origin::Multiply => Some(sized(
&mk,
"copy-overflow",
2,
&format!("copy length computed by multiplication (integer overflow?){dst}"),
p.bounded,
)),
_ => None,
}
}
}
}
fn sized(
mk: &dyn Fn(&'static str, u8, String) -> Finding,
pattern: &'static str,
severity: u8,
detail: &str,
bounded: bool,
) -> Finding {
if bounded {
mk(
pattern,
1,
format!("{detail}; but clamped or masked before the call, likely safe"),
)
} else {
mk(pattern, severity, detail.to_string())
}
}
fn arg_register(n: u8, win64: bool) -> Option<Register> {
let table: &[Register] = if win64 {
&[Register::RCX, Register::RDX, Register::R8, Register::R9]
} else {
&[
Register::RDI,
Register::RSI,
Register::RDX,
Register::RCX,
Register::R8,
Register::R9,
]
};
table.get((n as usize).checked_sub(1)?).copied()
}
struct Prov {
origin: Origin,
bounded: bool,
}
fn provenance(
an: &Analysis,
bin: &Binary,
bytes: &[u8],
block: &crate::analysis::engine::BasicBlock,
call_idx: usize,
arg: u8,
win64: bool,
) -> Prov {
let Some(target) = arg_register(arg, win64) else {
return Prov {
origin: Origin::Unknown,
bounded: false,
};
};
let mut want = target.full_register();
let mut bounded = false;
let mut info = InstructionInfoFactory::new();
for j in (0..call_idx).rev() {
let ins = &block.insns[j];
let Some(d) = decode_one(&ins.bytes, ins.addr, an.bits) else {
continue;
};
if matches!(d.mnemonic(), Mnemonic::Call) {
return Prov {
origin: Origin::Unknown,
bounded,
};
}
if is_cmov(&d) && writes_reg(&mut info, &d, want) {
bounded = true;
continue;
}
let writes = writes_reg(&mut info, &d, want);
if !writes {
continue;
}
if d.mnemonic() == Mnemonic::And && d.op1_kind() != OpKind::Register {
bounded = true;
}
if d.mnemonic() == Mnemonic::Mov
&& d.op0_kind() == OpKind::Register
&& d.op1_kind() == OpKind::Register
{
want = d.op1_register().full_register();
continue;
}
return Prov {
origin: origin_of(&d, bin, an, bytes),
bounded,
};
}
Prov {
origin: Origin::Unknown,
bounded,
}
}
fn writes_reg(info: &mut InstructionInfoFactory, d: &Instruction, want: Register) -> bool {
info.info(d)
.used_registers()
.iter()
.filter(|u| {
matches!(
u.access(),
OpAccess::Write | OpAccess::ReadWrite | OpAccess::CondWrite
)
})
.any(|u| u.register().is_gpr() && u.register().full_register() == want)
}
fn is_cmov(d: &Instruction) -> bool {
format!("{:?}", d.mnemonic()).starts_with("Cmov")
}
fn origin_of(d: &Instruction, bin: &Binary, an: &Analysis, bytes: &[u8]) -> Origin {
match d.mnemonic() {
Mnemonic::Imul | Mnemonic::Mul => Origin::Multiply,
Mnemonic::Shl | Mnemonic::Sal => Origin::Multiply,
Mnemonic::Sub | Mnemonic::Sbb => Origin::Subtract,
Mnemonic::Lea => {
let base = d.memory_base();
if matches!(
base,
Register::RSP | Register::ESP | Register::RBP | Register::EBP
) && d.memory_index() == Register::None
{
Origin::Stack
} else if d.memory_index() != Register::None {
Origin::Multiply
} else {
Origin::Fixed
}
}
Mnemonic::Mov => classify_mov(d, bin, an, bytes),
Mnemonic::Movzx | Mnemonic::Movsx | Mnemonic::Movsxd => Origin::Dynamic,
Mnemonic::Xor => {
if d.op0_register() == d.op1_register() {
Origin::Constant
} else {
Origin::Dynamic
}
}
_ => Origin::Dynamic,
}
}
fn classify_mov(d: &Instruction, bin: &Binary, an: &Analysis, bytes: &[u8]) -> Origin {
match d.op1_kind() {
OpKind::Immediate8
| OpKind::Immediate16
| OpKind::Immediate32
| OpKind::Immediate32to64
| OpKind::Immediate64 => {
let imm = d.immediate(1);
if points_at_string(bin, an, bytes, imm) {
Origin::Fixed
} else {
Origin::Constant
}
}
OpKind::Memory => Origin::Dynamic,
OpKind::Register => Origin::Dynamic,
_ => Origin::Dynamic,
}
}
fn points_at_string(bin: &Binary, an: &Analysis, bytes: &[u8], va: u64) -> bool {
let base = an.display_base;
let Some(off) = crate::analysis::engine::va_to_off(bin, base, va) else {
return false;
};
bytes
.get(off..off + 4)
.is_some_and(|w| w.iter().all(|&b| (0x20..0x7f).contains(&b)))
}
fn decode_one(raw: &[u8], ip: u64, bits: u32) -> Option<Instruction> {
if raw.is_empty() {
return None;
}
let mut dec = Decoder::with_ip(bits, raw, ip, DecoderOptions::NONE);
if !dec.can_decode() {
return None;
}
let insn = dec.decode();
(!insn.is_invalid()).then_some(insn)
}
fn reachable_functions(an: &Analysis, bin: &Binary) -> BTreeSet<u64> {
let base = crate::analysis::engine::display_base(bin);
let is_func: BTreeSet<u64> = an.functions.iter().map(|f| f.addr).collect();
let mut roots: Vec<u64> = bin
.symbols
.iter()
.filter(|s| s.kind == crate::model::SymKind::Export)
.map(|s| s.addr + base)
.filter(|a| is_func.contains(a))
.collect();
let entry = bin.entry + base;
if is_func.contains(&entry) {
roots.push(entry);
}
let mut seen: BTreeSet<u64> = roots.iter().copied().collect();
let mut queue: VecDeque<u64> = roots.into_iter().collect();
while let Some(a) = queue.pop_front() {
let Some(f) = an.find_function(a) else {
continue;
};
for &callee in &f.calls {
if is_func.contains(&callee) && seen.insert(callee) {
queue.push_back(callee);
}
}
}
seen
}
#[cfg(test)]
mod tests {
use super::*;
use crate::analysis::engine;
use crate::db::Db;
use crate::model::{Section, SymKind, Symbol};
struct Harness {
bin: Binary,
bytes: Vec<u8>,
}
impl Harness {
fn new(sink: &str, mut code: Vec<u8>) -> Harness {
let va = 0x1000u64;
let slot = 0x4000u64;
let call_at = va + code.len() as u64;
let disp = (slot as i64 - (call_at as i64 + 6)) as i32;
code.push(0xff);
code.push(0x15);
code.extend_from_slice(&disp.to_le_bytes());
code.push(0xc3);
let mut bin = Binary::stub(Format::Elf, Arch::X86_64);
bin.entry = va;
bin.sections = vec![Section {
name: ".text".into(),
vaddr: va,
vsize: code.len() as u64,
file_off: va,
file_size: code.len() as u64,
entropy: 0.0,
read: true,
write: false,
exec: true,
}];
bin.symbols = vec![Symbol {
addr: slot,
name: sink.into(),
kind: SymKind::Import,
}];
let mut bytes = vec![0u8; va as usize];
bytes.extend_from_slice(&code);
Harness { bin, bytes }
}
fn findings(&self) -> Vec<Finding> {
let an = engine::analyze(&self.bin, &self.bytes, 10_000, &Db::default());
run(&an, &self.bin, &self.bytes)
}
}
#[test]
fn memcpy_length_from_subtraction_is_flagged() {
let code = vec![
0x48, 0x89, 0xf7, 0x89, 0xca, 0x29, 0xc2, ];
let f = Harness::new("memcpy", code).findings();
let hit = f
.iter()
.find(|x| x.pattern == "copy-underflow")
.expect("expected copy-underflow");
assert_eq!(hit.severity, 3, "an unclamped subtraction is high severity");
}
#[test]
fn a_clamped_subtraction_is_downgraded_not_dropped() {
let code = vec![
0x29, 0xc2, 0x39, 0xca, 0x0f, 0x47, 0xd1, ];
let f = Harness::new("memcpy", code).findings();
let hit = f
.iter()
.find(|x| x.pattern == "copy-underflow")
.expect("still reported");
assert_eq!(hit.severity, 1, "a clamped subtraction is downgraded");
assert!(hit.detail.contains("clamped or masked"));
}
#[test]
fn a_clamp_through_a_move_hop_is_detected() {
let code = vec![
0x49, 0x29, 0xc4, 0x49, 0x39, 0xcc, 0x4c, 0x0f, 0x47, 0xe1, 0x4c, 0x89, 0xe2, ];
let f = Harness::new("memcpy", code).findings();
let hit = f
.iter()
.find(|x| x.pattern == "copy-underflow")
.expect("still reported");
assert_eq!(hit.severity, 1, "the clamp is seen through the move hop");
}
#[test]
fn strcpy_into_a_stack_buffer_is_a_stack_overflow() {
let code = vec![0x48, 0x8d, 0x7c, 0x24, 0x20]; let f = Harness::new("strcpy", code).findings();
let hit = f
.iter()
.find(|x| x.pattern == "stack-overflow")
.expect("expected stack-overflow");
assert_eq!(hit.severity, 3);
assert!(hit.detail.contains("stack buffer"));
}
#[test]
fn strcpy_into_a_heap_pointer_is_a_plain_unbounded_copy() {
let code = vec![0x48, 0x8b, 0x3b]; let f = Harness::new("strcpy", code).findings();
assert!(f.iter().any(|x| x.pattern == "unbounded-copy"));
assert!(!f.iter().any(|x| x.pattern == "stack-overflow"));
}
#[test]
fn malloc_size_from_multiply_is_flagged() {
let code = vec![
0x89, 0xf8, 0x0f, 0xaf, 0xc6, 0x89, 0xc7, ];
let f = Harness::new("malloc", code).findings();
assert!(
f.iter().any(|x| x.pattern == "alloc-overflow"),
"expected alloc-overflow, got {:?}",
f.iter().map(|x| x.pattern).collect::<Vec<_>>()
);
}
#[test]
fn memcpy_with_a_constant_length_is_not_flagged() {
let code = vec![0xba, 0x10, 0x00, 0x00, 0x00]; let f = Harness::new("memcpy", code).findings();
assert!(
!f.iter().any(|x| x.pattern.starts_with("copy-")),
"a constant length must not be flagged: {:?}",
f.iter().map(|x| x.pattern).collect::<Vec<_>>()
);
}
#[test]
fn printf_with_a_runtime_format_is_flagged() {
let code = vec![0x48, 0x8b, 0x3e]; let f = Harness::new("printf", code).findings();
assert!(
f.iter().any(|x| x.pattern == "format-string"),
"expected format-string, got {:?}",
f.iter().map(|x| x.pattern).collect::<Vec<_>>()
);
}
#[test]
fn printf_with_a_fixed_format_pointer_is_not_flagged() {
let code = vec![0x48, 0x8d, 0x3d, 0x10, 0x00, 0x00, 0x00]; let f = Harness::new("printf", code).findings();
assert!(
!f.iter().any(|x| x.pattern == "format-string"),
"a fixed format pointer must not be flagged: {:?}",
f.iter().map(|x| x.pattern).collect::<Vec<_>>()
);
}
#[test]
fn strcpy_reachable_from_entry_is_high_severity() {
let f = Harness::new("strcpy", Vec::new()).findings();
let u = f
.iter()
.find(|x| x.pattern == "unbounded-copy")
.expect("unbounded-copy expected");
assert!(u.reachable);
assert_eq!(u.severity, 3);
}
#[test]
fn a_value_set_before_an_intervening_call_is_not_attributed() {
let va = 0x1000u64;
let other = 0x5000u64;
let mut code = vec![0x29, 0xc2]; let call_at = va + code.len() as u64;
let disp = (other as i64 - (call_at as i64 + 6)) as i32;
code.push(0xff);
code.push(0x15);
code.extend_from_slice(&disp.to_le_bytes()); let mut h = Harness::new("memcpy", code);
h.bin.symbols.push(Symbol {
addr: other,
name: "getpid".into(),
kind: SymKind::Import,
});
let f = h.findings();
assert!(
!f.iter().any(|x| x.pattern == "copy-underflow"),
"a value set before an intervening call must not be attributed"
);
}
}