diff --git a/binaryninjaapi.h b/binaryninjaapi.h index d5e612cfcd..4eb55c7855 100644 --- a/binaryninjaapi.h +++ b/binaryninjaapi.h @@ -13574,10 +13574,31 @@ namespace BinaryNinja { \return The HLIL for this Function if it's available. */ Ref GetHighLevelILIfAvailable() const; + Ref GetLanguageRepresentation(const std::string& language = "Pseudo C") const; Ref GetLanguageRepresentationIfAvailable( const std::string& language = "Pseudo C") const; + // Code complexity primitives - thin wrappers over the matching BNGetFunction*Complexity + // Core calls, each a single whole-function walk. + double GetCyclomaticComplexity() const; + double GetInstructionCountComplexity() const; + double GetTokenCountComplexity() const; + double GetBranchDensityComplexity() const; + double GetHalsteadComplexity() const; + double GetNestingDepthComplexity() const; + double GetCognitiveComplexity() const; + + // Composite/aggregate complexity metrics - built entirely from the primitives above plus + // other already-public API calls (MLIL visitors, BinaryView::GetCodeReferences, + // BinaryView::GetAnalysisFunctionsForAddress). Implemented here rather than in Core since + // none of them need a per-instruction callback into Core - only a bounded number of + // whole-function calls - so there's nothing to gain from Core doing the aggregation itself. + double GetCompositeComplexity() const; + double GetFanOutComplexity() const; + double GetCodeReferenceComplexity() const; + double GetTransitiveComplexity() const; + Ref GetType() const; Confidence> GetReturnType() const; ReturnValue GetReturnValue() const; diff --git a/binaryninjacore.h b/binaryninjacore.h index a70f8e941a..8916dfea6b 100644 --- a/binaryninjacore.h +++ b/binaryninjacore.h @@ -6071,6 +6071,18 @@ extern "C" BINARYNINJACOREAPI BNLanguageRepresentationFunction* BNGetFunctionLanguageRepresentationIfAvailable( BNFunction* func, const char* language); + // Code complexity primitives - see core/complexity.cpp. Each is a single whole-function walk; + // composite/aggregate metrics built from these (composite, fan-out, code references, + // transitive) are implemented in the public API layer instead, on top of these and other + // already-public calls. + BINARYNINJACOREAPI double BNGetFunctionCyclomaticComplexity(BNFunction* func); + BINARYNINJACOREAPI double BNGetFunctionInstructionCountComplexity(BNFunction* func); + BINARYNINJACOREAPI double BNGetFunctionTokenCountComplexity(BNFunction* func); + BINARYNINJACOREAPI double BNGetFunctionBranchDensityComplexity(BNFunction* func); + BINARYNINJACOREAPI double BNGetFunctionHalsteadComplexity(BNFunction* func); + BINARYNINJACOREAPI double BNGetFunctionNestingDepthComplexity(BNFunction* func); + BINARYNINJACOREAPI double BNGetFunctionCognitiveComplexity(BNFunction* func); + BINARYNINJACOREAPI BNDataBuffer* BNGetConstantData(BNFunction* func, BNRegisterValueType state, uint64_t value, size_t size, BNBuiltinType* builtin); diff --git a/function.cpp b/function.cpp index d5dd33e839..54bab6d094 100644 --- a/function.cpp +++ b/function.cpp @@ -21,7 +21,10 @@ #include "binaryninjaapi.h" #include "mediumlevelilinstruction.h" #include "highlevelilinstruction.h" +#include #include +#include +#include #include "ffi.h" @@ -739,6 +742,209 @@ Ref Function::GetLanguageRepresentationIfAvailab } +double Function::GetCyclomaticComplexity() const { return BNGetFunctionCyclomaticComplexity(m_object); } +double Function::GetInstructionCountComplexity() const { return BNGetFunctionInstructionCountComplexity(m_object); } +double Function::GetTokenCountComplexity() const { return BNGetFunctionTokenCountComplexity(m_object); } +double Function::GetBranchDensityComplexity() const { return BNGetFunctionBranchDensityComplexity(m_object); } +double Function::GetHalsteadComplexity() const { return BNGetFunctionHalsteadComplexity(m_object); } +double Function::GetNestingDepthComplexity() const { return BNGetFunctionNestingDepthComplexity(m_object); } +double Function::GetCognitiveComplexity() const { return BNGetFunctionCognitiveComplexity(m_object); } + + +double Function::GetCompositeComplexity() const +{ + // Default weights used to blend the individual primitives into a single score. Tuned so that + // a small, simple function scores in the single digits and a deeply nested, branchy function + // scores in the hundreds. Callers who want different behavior should call the primitives + // directly and combine them with their own weights. + const double cyclomaticWeight = 0.25; + const double instructionCountWeight = 0.05; + const double branchDensityWeight = 10.0; + const double halsteadWeight = 0.05; + const double nestingDepthWeight = 2.0; + const double cognitiveWeight = 0.5; + + return cyclomaticWeight * GetCyclomaticComplexity() + + instructionCountWeight * GetInstructionCountComplexity() + + branchDensityWeight * GetBranchDensityComplexity() + + halsteadWeight * GetHalsteadComplexity() + + nestingDepthWeight * GetNestingDepthComplexity() + + cognitiveWeight * GetCognitiveComplexity(); +} + + +namespace { + + // MLIL call-like operations whose "dest" operand may resolve to a callee. + const unordered_set g_mlilCallOperations = { + MLIL_CALL, MLIL_CALL_UNTYPED, MLIL_CALL_SSA, MLIL_CALL_UNTYPED_SSA, + MLIL_TAILCALL, MLIL_TAILCALL_UNTYPED, MLIL_TAILCALL_SSA, MLIL_TAILCALL_UNTYPED_SSA + }; + + // Syscalls target a syscall number, not an address - always treated as "external" (opaque). + const unordered_set g_mlilSyscallOperations = { + MLIL_SYSCALL, MLIL_SYSCALL_SSA, MLIL_SYSCALL_UNTYPED, MLIL_SYSCALL_UNTYPED_SSA + }; + + // Symbol types whose implementation isn't in this binary (imports/thunks/library stubs) - we + // can see that a call goes there, but not what happens once it does. + const unordered_set g_externalSymbolTypes = { + ImportAddressSymbol, ImportedFunctionSymbol, ExternalSymbol, LibraryFunctionSymbol + }; + + struct CallSites + { + // One entry per call site that resolves to a function defined in this binary (duplicates + // included, e.g. the same helper called twice shows up twice). + vector> internalCallees; + size_t externalCount = 0; + size_t indirectCount = 0; + }; + + // Walks every call-like MLIL statement in `func` and sorts it into one of three buckets: + // + // * resolves to a function defined in this binary -> appended to internalCallees + // * resolves to an address, but that address is an import/thunk/library stub, or analysis + // couldn't identify a function there -> externalCount + // * the target isn't a static constant at all (computed/register call, e.g. through a + // function pointer or vtable) -> indirectCount, since we can't know what runs there without + // deeper (and possibly incomplete) analysis + // + // Built entirely from already-public API calls (MLIL visitors, BinaryView lookups) - this used + // to be Core-internal, but nothing here needs a per-instruction callback into Core, so there's + // nothing to gain from Core doing it on our behalf. + CallSites ClassifyCallSites(const Function* func) + { + CallSites result; + + Ref mlil = func->GetMediumLevelIL(); + if (!mlil) + return result; + + Ref view = func->GetView(); + + mlil->VisitInstructions([&](BasicBlock*, const MediumLevelILInstruction& instr) { + if (g_mlilSyscallOperations.count(instr.operation)) + { + result.externalCount++; + return; + } + if (!g_mlilCallOperations.count(instr.operation)) + return; + + bool foundDest = false; + MediumLevelILInstruction dest; + for (auto& operand : instr.GetOperands()) + { + if (operand.GetUsage() == DestExprMediumLevelOperandUsage && operand.GetType() == ExprMediumLevelOperand) + { + dest = operand.GetExpr(); + foundDest = true; + break; + } + } + + if (!foundDest || + (dest.operation != MLIL_CONST && dest.operation != MLIL_CONST_PTR && dest.operation != MLIL_IMPORT)) + { + result.indirectCount++; + return; + } + + uint64_t targetAddr = dest.GetRawOperandAsInteger(0); + vector> targets = view->GetAnalysisFunctionsForAddress(targetAddr); + if (targets.empty()) + { + result.externalCount++; + return; + } + + Ref symbol = view->GetSymbolByAddress(targetAddr); + if (symbol && g_externalSymbolTypes.count(symbol->GetType())) + result.externalCount++; + else + result.internalCallees.push_back(targets[0]); + }); + + return result; + } + + + // `func`'s own composite complexity plus the (decayed) composite complexity of everything it + // calls, recursively, up to maxDepth hops. Each function is only ever counted once (tracked by + // start address) across the whole walk, which both breaks recursion/cycles safely and avoids + // over-counting a shared helper reached through multiple paths. + double TransitiveComplexityWalk(const Function* func, size_t depth, double decay, unordered_set& visited) + { + if (visited.count(func->GetStart())) + return 0.0; + visited.insert(func->GetStart()); + + double total = func->GetCompositeComplexity(); + if (depth == 0) + return total; + + unordered_map> callees; + for (auto& callee : ClassifyCallSites(func).internalCallees) + callees.emplace(callee->GetStart(), callee); + + for (auto& entry : callees) + total += decay * TransitiveComplexityWalk(entry.second.GetPtr(), depth - 1, decay, visited); + + return total; + } + +} // namespace + + +// How much, and how riskily, `func` fans out into other code - captures the "main() is just a +// dispatcher" shape that every intraprocedural metric misses. Distinct callees count in full; +// repeated calls to a callee already counted (e.g. in a loop) count for less. Calls to +// external/library code count more, since we can't inspect what they do. Indirect calls count the +// most, since we can't even statically know what runs. +double Function::GetFanOutComplexity() const +{ + CallSites calls = ClassifyCallSites(this); + + unordered_set distinctTargets; + for (auto& callee : calls.internalCallees) + distinctTargets.insert(callee->GetStart()); + + size_t repeatCalls = calls.internalCallees.size() - distinctTargets.size(); + + return double(distinctTargets.size()) + + 0.25 * double(repeatCalls) + + 1.5 * double(calls.externalCount) + + 2.5 * double(calls.indirectCount); +} + + +// How many places in this binary's code reference `func` - the complement of GetFanOutComplexity: +// that measures how much `func` reaches into other code, this measures how much other code +// reaches into `func`. Not "complexity" in the same sense as the other metrics - a heavily- +// referenced function isn't necessarily hard to read on its own - but it's a direct measure of +// blast radius: a change to a function referenced from hundreds of call sites has far more to +// potentially break than one referenced from a single caller. +double Function::GetCodeReferenceComplexity() const +{ + return double(GetView()->GetCodeReferences(GetStart()).size()); +} + + +// `func`'s own composite complexity plus the (decayed) composite complexity of everything it +// calls, recursively, up to two hops. This is the metric that actually answers "this function +// looks simple, but how much do I have to read to understand what it does": a function with a low +// `composite` score but a high `transitive` score is exactly a thin dispatcher over a lot of real +// work. +double Function::GetTransitiveComplexity() const +{ + const size_t maxDepth = 2; + const double decay = 0.5; + unordered_set visited; + return TransitiveComplexityWalk(this, maxDepth, decay, visited); +} + + Ref Function::GetType() const { return new Type(BNGetFunctionType(m_object)); diff --git a/python/complexity.py b/python/complexity.py new file mode 100644 index 0000000000..e4eeabb1bb --- /dev/null +++ b/python/complexity.py @@ -0,0 +1,648 @@ +# coding=utf-8 +# Copyright (c) 2015-2026 Vector 35 Inc +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to +# deal in the Software without restriction, including without limitation the +# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or +# sell copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING +# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS +# IN THE SOFTWARE. + +""" +Code complexity metrics for :py:class:`~binaryninja.function.Function`. + +Each metric isolates a different factor that contributes to how hard a function is to read, +test, or reason about: + +============================ ===================================================================== +Metric Factor measured +============================ ===================================================================== +``cyclomatic`` Number of linearly independent paths through the control flow graph + (classic McCabe metric: edges - nodes + 2). +``instruction_count`` Raw code length, counted in MLIL instructions (including sub-expressions). +``token_count`` Raw lexical size: total disassembly text tokens (mnemonics, + operands, punctuation) - the literal size of the listing you'd + read, as opposed to `instruction_count`'s semantic MLIL node count. +``branch_density`` Fraction of top-level MLIL statements that alter control flow. +``halstead`` Diversity of instruction "vocabulary" (distinct operators/operands), + using Halstead's volume ``V = N * log2(n)``. +``nesting_depth`` Maximum nesting depth of structured control flow (if/while/for/switch) + in the HLIL AST. +``cognitive`` SonarSource-style cognitive complexity: control flow breaks cost more + the more deeply they are nested. +``composite`` Weighted blend of all of the above into a single score. +``fan_out`` Number/riskiness of the calls this function makes (one hop only). +``code_references`` Number of places in the binary's code that reference this function - + the complement of `fan_out`: blast radius, not internal complexity. +``transitive`` Composite complexity of this function plus everything it calls, + recursively - "how much do I actually have to read". +============================ ===================================================================== + +Every metric from `fan_out` on is interprocedural: it looks beyond the function's own body. That +misses a common real-world shape - a `main()` that is just a flat sequence of calls to a dozen +helpers looks trivial by every intraprocedural metric (low cyclomatic complexity, no nesting) even +though understanding it fully means reading everything it calls. `fan_out`, `code_references`, and +`transitive` capture that (`code_references` in the opposite direction: callers, not callees). + +All metrics are exposed through a single entry point, :py:func:`get_code_complexity`, which is +also available as :py:meth:`Function.get_complexity() `. +""" + +from typing import Callable, Dict, Iterable, Iterator, List, NamedTuple, Optional, Set, Tuple, TYPE_CHECKING, Union + +from . import _binaryninjacore as core +from . import mediumlevelil +from .enums import MediumLevelILOperation, SymbolType +from .log import log_warn + +if TYPE_CHECKING: + from . import function + + +FunctionKey = Tuple[str, str, int] + + +def function_key(func: 'function.Function') -> FunctionKey: + """Return a stable identity for a function, including the architecture/platform at its address.""" + platform = getattr(func, 'platform', None) + arch = getattr(func, 'arch', None) + return ( + getattr(platform, 'name', '') if platform is not None else '', + getattr(arch, 'name', '') if arch is not None else '', + int(func.start), + ) + + +_MLIL_CALL_OPERATIONS = { + MediumLevelILOperation.MLIL_CALL, + MediumLevelILOperation.MLIL_CALL_UNTYPED, + MediumLevelILOperation.MLIL_CALL_SSA, + MediumLevelILOperation.MLIL_CALL_UNTYPED_SSA, + MediumLevelILOperation.MLIL_TAILCALL, + MediumLevelILOperation.MLIL_TAILCALL_UNTYPED, + MediumLevelILOperation.MLIL_TAILCALL_SSA, + MediumLevelILOperation.MLIL_TAILCALL_UNTYPED_SSA, +} + +_MLIL_SYSCALL_OPERATIONS = { + MediumLevelILOperation.MLIL_SYSCALL, + MediumLevelILOperation.MLIL_SYSCALL_SSA, + MediumLevelILOperation.MLIL_SYSCALL_UNTYPED, + MediumLevelILOperation.MLIL_SYSCALL_UNTYPED_SSA, +} + +# Symbol types whose implementation isn't in this binary (imports/thunks/library stubs) - we can +# see that a call goes there, but not what happens once it does. +_EXTERNAL_SYMBOL_TYPES = { + SymbolType.ImportAddressSymbol, + SymbolType.ImportedFunctionSymbol, + SymbolType.ExternalSymbol, + SymbolType.LibraryFunctionSymbol, +} + +# Default weights used to blend the individual metrics into `composite`. These are tuned so that +# a "typical" small function scores roughly in the single digits and a deeply nested, branchy +# function scores in the tens to hundreds. Callers who want different behavior should call the +# individual metrics directly and combine them with their own weights. +DEFAULT_COMPOSITE_WEIGHTS: Dict[str, float] = { + 'cyclomatic': 0.25, + 'instruction_count': 0.05, + 'branch_density': 10.0, + 'halstead': 0.05, + 'nesting_depth': 2.0, + 'cognitive': 0.5, +} + + +def cyclomatic_complexity(func: 'function.Function') -> float: + """ + McCabe cyclomatic complexity: ``M = E - N + 2`` over the function's basic block graph. + + Measures the number of branches in the control flow graph, independent of function size or + the kinds of instructions used. Computed natively in Core - see + :py:meth:`~binaryninja.function.Function.get_complexity`. + """ + return core.BNGetFunctionCyclomaticComplexity(func.handle) + + +def instruction_count_complexity(func: 'function.Function') -> float: + """ + Raw code length: total number of MLIL instructions, including sub-expressions. 0.0 if MLIL is + not available for this function. Computed natively in Core. + """ + return core.BNGetFunctionInstructionCountComplexity(func.handle) + + +def token_count_complexity(func: 'function.Function') -> float: + """ + Raw lexical size: total number of disassembly text tokens (mnemonics, operands, punctuation, + etc.) across the function. + + Unlike `instruction_count`, which counts semantic MLIL nodes, this counts the literal tokens + of the disassembly listing you'd actually read - two functions with the same MLIL shape can + still differ here if one uses instructions with more operands/addressing-mode tokens per line + than the other. Always available, even for functions without MLIL/HLIL. Computed natively in + Core. + """ + return core.BNGetFunctionTokenCountComplexity(func.handle) + + +def branch_density_complexity(func: 'function.Function') -> float: + """ + Fraction (0.0-1.0) of top-level MLIL statements that alter control flow (if/goto/jump/tailcall). + + Two functions can have the same instruction count but very different "branchiness"; this + metric isolates that factor independent of raw length. 0.0 if MLIL is not available for this + function. Computed natively in Core. + """ + return core.BNGetFunctionBranchDensityComplexity(func.handle) + + +def halstead_volume_complexity(func: 'function.Function') -> float: + """ + Halstead volume ``V = N * log2(n)`` over MLIL, where: + + * ``n`` (vocabulary) is the number of distinct operators (MLIL operation kinds) plus the + number of distinct operands (constants/variables referenced). + * ``N`` (length) is the total number of operator and operand occurrences. + + This captures how diverse the instruction "types" used by a function are, as distinct from + how many instructions there are in total. 0.0 if MLIL is not available for this function. + Computed natively in Core. + """ + return core.BNGetFunctionHalsteadComplexity(func.handle) + + +def nesting_depth_complexity(func: 'function.Function') -> float: + """ + Maximum nesting depth of structured control flow (if/while/for/do-while/switch) in the HLIL AST. + + Two functions with identical cyclomatic complexity can differ a lot in how deeply their + branches are nested; deeply nested code is harder to hold in your head even with the same + number of paths. 0.0 if HLIL is not available for this function. Computed natively in Core. + """ + return core.BNGetFunctionNestingDepthComplexity(func.handle) + + +def cognitive_complexity(func: 'function.Function') -> float: + """ + Simplified SonarSource-style cognitive complexity over the HLIL AST. + + Every control-flow-breaking construct (if/while/for/do-while/switch) costs 1 point, plus 1 + additional point for every level of nesting it sits inside. Unlike cyclomatic complexity, + this penalizes deeply nested logic more than the same number of branches laid out flat. 0.0 if + HLIL is not available for this function. Computed natively in Core. + """ + return core.BNGetFunctionCognitiveComplexity(func.handle) + + +class ComplexityCache: + """ + Memoizes the per-function work that :py:func:`composite_complexity`, :py:func:`fan_out_complexity`, + and :py:func:`transitive_complexity` do, keyed by platform, architecture, and start address. + + A single function is very often reachable from many different callers (a shared helper, a + logging routine, an allocator wrapper). Without this, scoring every function in a binary via + :py:func:`transitive_complexity` would redo that shared helper's full analysis - CFG walk, + MLIL/HLIL traversal, its own call classification - once per caller that reaches it, instead of + once total. Pass the same cache instance across every function in a scan (see + :py:func:`compute_all_complexity`) to get that reuse; the default of not passing one at all + keeps a single one-off call to any of these functions correct (just not sped up by anything + computed elsewhere). + """ + + def __init__(self, functions: Optional[Iterable['function.Function']] = None): + self.composite: Dict[FunctionKey, float] = {} + self.call_sites: Dict[FunctionKey, '_CallSites'] = {} + self.functions: Dict[FunctionKey, 'function.Function'] = {} + self.functions_by_address: Dict[int, List['function.Function']] = {} + if functions is not None: + for func in functions: + self.remember_function(func) + + def remember_function(self, func: 'function.Function') -> None: + key = function_key(func) + self.functions[key] = func + bucket = self.functions_by_address.setdefault(func.start, []) + if all(function_key(existing) != key for existing in bucket): + bucket.append(func) + + def forget_function(self, key: FunctionKey) -> None: + func = self.functions.pop(key, None) + self.composite.pop(key, None) + self.call_sites.pop(key, None) + if func is None: + return + bucket = self.functions_by_address.get(func.start) + if bucket is None: + return + self.functions_by_address[func.start] = [candidate for candidate in bucket if function_key(candidate) != key] + if not self.functions_by_address[func.start]: + del self.functions_by_address[func.start] + + def resolve_function(self, caller: 'function.Function', addr: int) -> Optional['function.Function']: + """Resolve an address without collapsing same-address functions from different platforms.""" + candidates = self.functions_by_address.get(addr) + if candidates is None: + candidates = list(caller.view.get_functions_at(addr)) + for candidate in candidates: + self.remember_function(candidate) + if not candidates: + return None + caller_platform = getattr(caller, 'platform', None) + caller_arch = getattr(caller, 'arch', None) + for candidate in candidates: + if getattr(candidate, 'platform', None) == caller_platform: + return candidate + for candidate in candidates: + if getattr(candidate, 'arch', None) == caller_arch: + return candidate + return candidates[0] + + +def composite_from_metrics(values: Dict[str, float], weights: Dict[str, float] = None) -> float: + """Combine already-computed component metrics without walking the function's IL again.""" + effective_weights = dict(DEFAULT_COMPOSITE_WEIGHTS) + if weights: + effective_weights.update(weights) + return sum(effective_weights[name] * values[name] for name in DEFAULT_COMPOSITE_WEIGHTS) + + +def composite_complexity( + func: 'function.Function', weights: Dict[str, float] = None, cache: ComplexityCache = None +) -> float: + """ + Weighted blend of every other metric into a single score. + + :param weights: Optional override for :py:data:`DEFAULT_COMPOSITE_WEIGHTS`. Any metric name + left out keeps its default weight. + :param cache: See :py:class:`ComplexityCache`. Only used when `weights` is left at its default, + since a cached value can't reflect caller-supplied weights. + """ + key = function_key(func) + if weights is None and cache is not None and key in cache.composite: + return cache.composite[key] + + result = composite_from_metrics({ + 'cyclomatic': cyclomatic_complexity(func), + 'instruction_count': instruction_count_complexity(func), + 'branch_density': branch_density_complexity(func), + 'halstead': halstead_volume_complexity(func), + 'nesting_depth': nesting_depth_complexity(func), + 'cognitive': cognitive_complexity(func), + }, weights) + + if weights is None and cache is not None: + cache.composite[key] = result + return result + + +class _CallSites(NamedTuple): + # Stable key of each call target that resolves to a function defined in this binary (duplicates + # included, e.g. the same helper called twice appears twice). Deliberately plain tuples rather + # than Function objects: this is what gets cached in ComplexityCache, and a + # Function wrapper keeps its underlying core analysis data alive for as long as anything holds + # a reference to it. A scan over every function in a large binary would otherwise end up + # pinning most of the binary's functions in memory for the scan's whole duration, just because + # each one was somebody's callee at some point. Including platform/architecture in the key also + # prevents same-address functions in multi-platform views from being silently conflated. + internal_callee_keys: List[FunctionKey] + external_count: int + indirect_count: int + + +def _classify_call_sites(func: 'function.Function', cache: ComplexityCache = None) -> _CallSites: + """ + Walks every call-like MLIL statement in `func` and sorts it into one of three buckets: + + * resolves to a function defined in this binary -> that function's stable key is appended to + `internal_callee_keys` + * resolves to an address, but that address is an import/thunk/library stub, or analysis + couldn't identify a function there -> `external_count` + * the target isn't a static constant at all (computed/register call, e.g. through a function + pointer or vtable) -> `indirect_count`, since we can't know what runs there without deeper + (and possibly incomplete) analysis + """ + key = function_key(func) + if cache is not None and key in cache.call_sites: + return cache.call_sites[key] + + mlil = func.mlil + if mlil is None: + result = _CallSites([], 0, 0) + if cache is not None: + cache.call_sites[key] = result + return result + + internal_callee_keys: List[FunctionKey] = [] + external_count = 0 + indirect_count = 0 + + for stmt in mlil.instructions: + if stmt.operation in _MLIL_SYSCALL_OPERATIONS: + external_count += 1 + continue + if stmt.operation not in _MLIL_CALL_OPERATIONS: + continue + + dest = stmt.dest + if not isinstance(dest, mediumlevelil.MediumLevelILConstBase): + indirect_count += 1 + continue + + target = ( + cache.resolve_function(func, dest.constant) + if cache is not None else func.view.get_function_at(dest.constant) + ) + if target is None: + external_count += 1 + continue + + symbol = func.view.get_symbol_at(dest.constant) + if symbol is not None and symbol.type in _EXTERNAL_SYMBOL_TYPES: + external_count += 1 + else: + internal_callee_keys.append(function_key(target)) + + result = _CallSites(internal_callee_keys, external_count, indirect_count) + if cache is not None: + cache.call_sites[key] = result + return result + + +def compute_metric_bundle( + func: 'function.Function', cache: ComplexityCache = None +) -> Dict[str, float]: + """Compute every non-transitive metric for `func` in one call. + + This is the fast path for table sweeps. The individual metric functions remain the public, + one-metric-at-a-time API; this just avoids recomputing every composite component and classifying + call sites more than once. The 7 non-interprocedural metrics are each a single native call into + Core, not a Python-side IL walk, so there's no separate traversal to batch here anymore - only + `fan_out`'s call-site classification is still Python-side work worth sharing with `cache`. + """ + if cache is None: + cache = ComplexityCache([func]) + else: + cache.remember_function(func) + key = function_key(func) + + calls = _classify_call_sites(func, cache) + distinct_targets = len(set(calls.internal_callee_keys)) + + values = { + 'cyclomatic': cyclomatic_complexity(func), + 'instruction_count': instruction_count_complexity(func), + 'token_count': token_count_complexity(func), + 'branch_density': branch_density_complexity(func), + 'halstead': halstead_volume_complexity(func), + 'nesting_depth': nesting_depth_complexity(func), + 'cognitive': cognitive_complexity(func), + 'fan_out': float( + distinct_targets + 0.25 * (len(calls.internal_callee_keys) - distinct_targets) + + 1.5 * calls.external_count + 2.5 * calls.indirect_count + ), + 'code_references': code_reference_count_complexity(func), + } + values['composite'] = composite_from_metrics(values) + cache.composite[key] = values['composite'] + return values + + +def fan_out_complexity(func: 'function.Function', cache: ComplexityCache = None) -> float: + """ + How much, and how riskily, `func` fans out into other code - captures the "`main()` is just a + dispatcher" shape that every intraprocedural metric misses. + + Distinct callees count in full; repeated calls to a callee already counted (e.g. in a loop) + count for less, since re-calling something you've already accounted for isn't new complexity. + Calls to external/library code count more, since you can't inspect what they do. Indirect + calls (through a computed target - function pointers, vtables, callbacks) count the most, + since you can't even statically know what runs. + + :param cache: See :py:class:`ComplexityCache`. + """ + calls = _classify_call_sites(func, cache) + distinct_targets = set(calls.internal_callee_keys) + repeat_calls = len(calls.internal_callee_keys) - len(distinct_targets) + return ( + len(distinct_targets) + + 0.25 * repeat_calls + + 1.5 * calls.external_count + + 2.5 * calls.indirect_count + ) + + +def code_reference_count_complexity(func: 'function.Function') -> int: + """ + How many places in this binary's code reference `func` - calls, tail calls, or any other code + cross-reference to its start address (see :py:func:`~binaryninja.binaryview.BinaryView.get_code_refs`). + The complement of `fan_out`: `fan_out` measures how much `func` reaches into other code, this + measures how much other code reaches into `func`. + + Not "complexity" in the same sense as the other metrics here - a heavily-referenced function + isn't necessarily hard to read on its own - but it is a direct measure of blast radius: a change + to a function referenced from hundreds of call sites has far more to potentially break than one + referenced from a single caller, which is exactly the kind of risk this report is for surfacing. + + This is a plain reverse-reference lookup against Binary Ninja's own cross-reference index, not a + walk that touches any other function's analysis - unlike `transitive`, computing this can't itself + trigger more analysis elsewhere, so it's always safe to include in an incremental recompute. + """ + return sum(1 for _ in func.view.get_code_refs(func.start)) + + +def transitive_complexity( + func: 'function.Function', + max_depth: int = 2, + decay: float = 0.5, + cache: ComplexityCache = None, + is_cancelled: Optional[Callable[[], bool]] = None, +) -> float: + """ + `func`'s own :py:func:`composite_complexity` plus the (decayed) composite complexity of + everything it calls, recursively, up to `max_depth` hops. + + This is the metric that actually answers "this function looks simple, but how much do I have + to read to understand what it does": a `main()` with a low `composite` score but a high + `transitive` score is exactly a thin dispatcher over a lot of real work. + + Each function is only ever counted once per call (tracked by stable function key), which both + breaks recursion/cycles safely and avoids over-counting a shared helper reached through + multiple paths - once you've read it, you don't need to read it again for a sibling call. Note + that this `visited` guard is intentionally local to a single top-level call (it exists to stop + double-counting *within* func's own subtree, e.g. two of its callees sharing a common + descendant) - it must not be shared across different top-level functions in a scan, or a + function reached by an earlier scanned function would wrongly read as contributing 0 to a + later, unrelated one. Sharing work *safely* across top-level calls in a scan is what `cache` + is for: see :py:class:`ComplexityCache`. + + :param max_depth: How many call hops to follow. Kept shallow by default since this walks the + call graph and cost grows with fan-out at each level. + :param decay: Weight applied per hop, so distant callees contribute less than the function's + own body. + :param cache: See :py:class:`ComplexityCache`. Pass the same instance across every function in + a whole-binary scan so a shared callee's own complexity is computed once, not + once per caller that reaches it. + :param is_cancelled: Optional callable polled throughout the recursive walk (not just once per + top-level call) so a caller running this on a background task can bail out + of a single expensive call promptly instead of waiting for the whole + walk - which, for a function embedded in a large, tightly-interconnected + cluster (e.g. a parser's mutually-calling helpers), can otherwise take a + very long time to unwind on its own. + """ + visited: Set[FunctionKey] = set() + view = func.view + + def walk(f: 'function.Function', depth: int) -> float: + if is_cancelled is not None and is_cancelled(): + return 0.0 + key = function_key(f) + if key in visited: + return 0.0 + visited.add(key) + + total = composite_complexity(f, cache=cache) + if depth <= 0: + return total + + callee_keys = set(_classify_call_sites(f, cache).internal_callee_keys) + for callee_key in callee_keys: + if is_cancelled is not None and is_cancelled(): + break + if callee_key in visited: + continue + callee = cache.functions.get(callee_key) if cache is not None else None + if callee is None: + candidates = list(view.get_functions_at(callee_key[2])) + callee = next((candidate for candidate in candidates if function_key(candidate) == callee_key), None) + if callee is not None: + total += decay * walk(callee, depth - 1) + return total + + return walk(func, max_depth) + + +# Single dispatch table backing the one public API function, `get_code_complexity`. +_METRICS: Dict[str, Callable[['function.Function'], float]] = { + 'cyclomatic': cyclomatic_complexity, + 'instruction_count': instruction_count_complexity, + 'token_count': token_count_complexity, + 'branch_density': branch_density_complexity, + 'halstead': halstead_volume_complexity, + 'nesting_depth': nesting_depth_complexity, + 'cognitive': cognitive_complexity, + 'composite': composite_complexity, + 'fan_out': fan_out_complexity, + 'code_references': code_reference_count_complexity, + 'transitive': transitive_complexity, +} + + +def list_complexity_metrics() -> List[str]: + """Names accepted by :py:func:`get_code_complexity`.""" + return list(_METRICS.keys()) + + +# Metrics whose computation can reuse a ComplexityCache across multiple functions. +_CACHEABLE_METRICS = {'composite', 'fan_out', 'transitive'} + + +def get_code_complexity( + func: 'function.Function', + metric: str = 'composite', + cache: ComplexityCache = None, + is_cancelled: Optional[Callable[[], bool]] = None, +) -> Union[int, float]: + """ + Compute a code complexity score for `func` using the named metric. + + :param func: Function to analyze. + :param metric: One of :py:func:`list_complexity_metrics` (default ``"composite"``). + :param cache: See :py:class:`ComplexityCache`. Only used by the interprocedural metrics + (``composite``, ``fan_out``, ``transitive``); ignored otherwise. Scanning many + functions should generally use :py:func:`compute_all_complexity` instead, which + sets this up for you. + :param is_cancelled: Optional callable polled during ``transitive``'s recursive call-graph walk + (the one metric whose single computation can itself take a long time on a + large, tightly-interconnected function); ignored by every other metric. + :raises ValueError: If `metric` is not a recognized metric name. + """ + try: + metric_fn = _METRICS[metric] + except KeyError: + raise ValueError(f"Unknown complexity metric {metric!r}; valid options are {list_complexity_metrics()}") + if metric == 'code_references': + return int(metric_fn(func)) + if metric == 'transitive': + return float(metric_fn(func, cache=cache, is_cancelled=is_cancelled)) + if metric in _CACHEABLE_METRICS: + return float(metric_fn(func, cache=cache)) + return float(metric_fn(func)) + + +def compute_all_complexity( + functions: 'Iterator[function.Function]', metrics: List[str] = None +) -> List[Tuple['function.Function', Dict[str, float]]]: + """ + Compute every metric in `metrics` for every function in `functions`, sharing one + :py:class:`ComplexityCache` across the whole batch. + + This is the efficient way to score every function in a binary: a shared helper reachable from + many different functions (a logging routine, an allocator wrapper, ...) has its own complexity + computed once here, no matter how many of the scanned functions call it - calling + :py:func:`get_code_complexity` separately per function in a loop would instead redo that + shared helper's analysis once per caller that reaches it within ``transitive``'s hop limit, + which is the dominant cost on a large binary with any commonly-used utility functions. + + :param functions: Functions to score, e.g. ``bv.functions``. + :param metrics: Which metrics to compute for each function (default: all of them). + :return: A list of ``(function, {metric_name: value})``, in the same order as `functions`. + """ + if metrics is None: + metrics = list_complexity_metrics() + functions = list(functions) + cache = ComplexityCache(functions) + results = [] + first_pass_metrics = [metric for metric in metrics if metric != 'transitive'] + for func in functions: + try: + if (set(_METRICS) - {'transitive'}).issubset(first_pass_metrics): + bundle = compute_metric_bundle(func, cache=cache) + values = {metric: bundle[metric] for metric in first_pass_metrics} + else: + values = {} + for metric in first_pass_metrics: + metric_fn = _METRICS[metric] + if metric == 'composite' and all(name in values for name in DEFAULT_COMPOSITE_WEIGHTS): + values[metric] = float(composite_from_metrics(values)) + cache.composite[function_key(func)] = values[metric] + elif metric == 'code_references': + values[metric] = int(metric_fn(func)) + else: + values[metric] = ( + float(metric_fn(func, cache=cache)) if metric in _CACHEABLE_METRICS else float(metric_fn(func)) + ) + except Exception as e: + log_warn(f'compute_all_complexity: skipping {func.name} at {func.start:#x}: {e}') + continue + results.append((func, values)) + if 'transitive' in metrics: + for func, values in results: + try: + values['transitive'] = float(transitive_complexity(func, cache=cache)) + except Exception as e: + log_warn(f'compute_all_complexity: skipping transitive for {func.name} at {func.start:#x}: {e}') + values['transitive'] = 0.0 + return results diff --git a/python/examples/complexity_report.py b/python/examples/complexity_report.py new file mode 100644 index 0000000000..f909037ba6 --- /dev/null +++ b/python/examples/complexity_report.py @@ -0,0 +1,178 @@ +# Copyright (c) 2015-2026 Vector 35 Inc +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to +# deal in the Software without restriction, including without limitation the +# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or +# sell copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING +# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS +# IN THE SOFTWARE. + +""" +Code Complexity Report +======================= + +An example plugin that surfaces ``Function.get_complexity()`` in the UI (and headlessly). + +Installed as a plugin, it registers two commands: + +* Right-click a function -> ``Plugins\\Code Complexity\\Show Report for This Function`` + Shows every complexity metric for the clicked function side by side, with a simple bar + visualization so the relative weight of each factor (length, branching, instruction + diversity, nesting) is easy to see at a glance. + +* ``Plugins\\Code Complexity\\Show Report for All Functions`` + Ranks every function in the binary by composite complexity and shows the top functions + with a full metric breakdown for each - a quick way to find the functions most worth + digging into or refactoring. + +It can also be run directly, headlessly, against a file on disk: + + python3 complexity_report.py /bin/ls +""" + +import sys +from typing import List + +import binaryninja +from binaryninja import complexity +from binaryninja.binaryview import BinaryView +from binaryninja.function import Function +from binaryninja.plugin import PluginCommand +from binaryninja.interaction import show_markdown_report + +METRICS = Function.complexity_metrics() + +# Rough "this is a lot" reference point for each metric, used only to scale the bar charts in +# the report - it has no effect on the scores themselves. Chosen from the distribution of +# values across the functions in a handful of real-world binaries. +_BAR_SCALE = { + 'cyclomatic': 30.0, + 'instruction_count': 400.0, + 'token_count': 1200.0, + 'branch_density': 0.5, + 'halstead': 8000.0, + 'nesting_depth': 8.0, + 'cognitive': 100.0, + 'composite': 300.0, + 'fan_out': 20.0, + 'code_references': 100.0, + 'transitive': 500.0, +} + +_METRIC_DESCRIPTIONS = { + 'cyclomatic': 'independent paths through the CFG.', + 'instruction_count': 'raw MLIL length.', + 'token_count': 'raw disassembly text token count.', + 'branch_density': 'fraction of statements that branch.', + 'halstead': 'volume from operator/operand diversity.', + 'nesting_depth': 'deepest if/while/for/switch nesting.', + 'cognitive': 'nesting-weighted control-flow cost.', + 'composite': 'weighted blend of the metrics above.', + 'fan_out': 'how much (and how riskily) this function calls out to other code.', + 'code_references': 'how many places in the code call/reference this function.', + 'transitive': "this function's composite complexity plus everything it calls, recursively.", +} + +_BAR_WIDTH = 20 + + +def _bar(metric: str, value: float) -> str: + scale = _BAR_SCALE.get(metric, 1.0) or 1.0 + filled = int(min(1.0, value / scale) * _BAR_WIDTH) + return '`' + ('#' * filled) + ('.' * (_BAR_WIDTH - filled)) + '`' + + +def _function_report_markdown(func: Function) -> str: + lines = [ + f'# Code Complexity: `{func.name}`', + '', + f'*{func.start:#x} in {func.view.file.filename}*', + '', + '| Metric | Value | Relative |', + '|---|---|---|', + ] + for metric in METRICS: + value = func.get_complexity(metric) + lines.append(f'| {metric} | {value:.2f} | {_bar(metric, value)} |') + lines.append('') + for metric in METRICS: + lines.append(f'- `{metric}`: {_METRIC_DESCRIPTIONS[metric]}') + return '\n'.join(lines) + + +def _binary_report_markdown(bv: BinaryView, top_n: int = 25) -> str: + # One shared cache for the whole scan: a helper reachable from many callers (logging, malloc + # wrappers, ...) gets its own complexity computed once here rather than once per caller that + # reaches it - see complexity.ComplexityCache. Without this, `transitive` in particular would + # redo the same common callees' analysis over and over on a binary with much shared code. + cache = complexity.ComplexityCache() + scored = [] + for func in bv.functions: + try: + scored.append((func, {m: complexity.get_code_complexity(func, m, cache=cache) for m in METRICS})) + except Exception as e: + binaryninja.log_warn(f'complexity_report: skipping {func.name}: {e}') + scored.sort(key=lambda entry: entry[1]['composite'], reverse=True) + + lines = [ + f'# Code Complexity Report: {bv.file.filename}', + '', + f'{len(scored)} functions analyzed. Showing the top {min(top_n, len(scored))} by composite score.', + '', + '| Function | ' + ' | '.join(METRICS) + ' |', + '|---|' + '---|' * len(METRICS), + ] + for func, values in scored[:top_n]: + row = ' | '.join(f'{values[m]:.2f}' for m in METRICS) + lines.append(f'| [{func.name}]({func.start:#x}) | {row} |') + return '\n'.join(lines) + + +def show_function_complexity(bv: BinaryView, func: Function) -> None: + show_markdown_report( + f'Code Complexity: {func.name}', _function_report_markdown(func), _function_report_markdown(func) + ) + + +def show_binary_complexity(bv: BinaryView) -> None: + show_markdown_report( + f'Code Complexity: {bv.file.filename}', _binary_report_markdown(bv), _binary_report_markdown(bv) + ) + + +PluginCommand.register_for_function( + 'Code Complexity\\Show Report for This Function', + 'Show every code complexity metric for the current function', + show_function_complexity, +) + +PluginCommand.register( + 'Code Complexity\\Show Report for All Functions', + 'Rank every function in the binary by code complexity', + show_binary_complexity, +) + + +if __name__ == '__main__': + if len(sys.argv) != 2: + print(f'usage: {sys.argv[0]} ') + sys.exit(1) + + bv = binaryninja.load(sys.argv[1]) + if bv is None: + print(f'could not load {sys.argv[1]}') + sys.exit(1) + bv.update_analysis_and_wait() + + print(_binary_report_markdown(bv)) diff --git a/python/examples/log_function_complexity.py b/python/examples/log_function_complexity.py new file mode 100644 index 0000000000..8284630434 --- /dev/null +++ b/python/examples/log_function_complexity.py @@ -0,0 +1,127 @@ +# Copyright (c) 2015-2026 Vector 35 Inc +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to +# deal in the Software without restriction, including without limitation the +# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or +# sell copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING +# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS +# IN THE SOFTWARE. + +""" +log_function_complexity.py +=========================== + +Headless script that loads a binary, computes every code complexity metric +(`Function.get_complexity()`) for every function, logs the results, and optionally writes them +to a CSV file for further analysis (e.g. in a spreadsheet, or to diff two versions of a binary). + +Usage: + + python3 log_function_complexity.py [--metric NAME] [--top N] [--csv out.csv] [--quiet] + +Examples: + + # Log every metric for every function, sorted by composite complexity + python3 log_function_complexity.py /bin/ls + + # Only the 10 most cyclomatically complex functions + python3 log_function_complexity.py /bin/ls --metric cyclomatic --top 10 + + # Everything, plus a CSV for spreadsheet analysis + python3 log_function_complexity.py /bin/ls --csv ls_complexity.csv +""" + +import argparse +import csv +import sys + +import binaryninja +from binaryninja import complexity +from binaryninja.function import Function + +METRICS = Function.complexity_metrics() + + +def parse_args(argv): + parser = argparse.ArgumentParser(description='Log code complexity metrics for every function in a binary.') + parser.add_argument('binary', help='Path to the binary to analyze') + parser.add_argument( + '--metric', choices=METRICS + ['all'], default='all', + help='Which metric to sort/log by (default: all metrics, sorted by composite)' + ) + parser.add_argument('--top', type=int, default=None, help='Only log the top N functions (default: all)') + parser.add_argument('--csv', default=None, help='Optional path to write a CSV of all metrics for all functions') + parser.add_argument('--quiet', action='store_true', help="Don't print a summary table to stdout") + return parser.parse_args(argv) + + +def compute_all(bv) -> list: + """ + Returns a list of (Function, {metric_name: value}) for every function that could be analyzed. + + Uses `complexity.compute_all_complexity`, which shares one `ComplexityCache` across the whole + binary: a helper reachable from many callers (logging, malloc wrappers, ...) has its own + complexity computed once for the scan, not once per caller that reaches it. Without that, the + `transitive` metric in particular would redo the same common callees' analysis over and over on + a binary with much shared code. + """ + return complexity.compute_all_complexity(bv.functions, METRICS) + + +def main(argv=None) -> int: + args = parse_args(sys.argv[1:] if argv is None else argv) + + bv = binaryninja.load(args.binary) + if bv is None: + binaryninja.log_error(f'log_function_complexity: could not load {args.binary}') + return 1 + bv.update_analysis_and_wait() + + sort_key = 'composite' if args.metric == 'all' else args.metric + results = compute_all(bv) + results.sort(key=lambda entry: entry[1][sort_key], reverse=True) + if args.top is not None: + results = results[:args.top] + + binaryninja.log_info( + f'log_function_complexity: analyzed {len(results)} function(s) in {args.binary}, sorted by {sort_key}' + ) + for func, values in results: + if args.metric == 'all': + detail = ', '.join(f'{m}={values[m]:.2f}' for m in METRICS) + else: + detail = f'{args.metric}={values[args.metric]:.2f}' + binaryninja.log_info(f'{func.name} @ {func.start:#x}: {detail}') + + if not args.quiet: + header = ['function', 'address'] + METRICS + widths = [max(len(header[i]), 10) for i in range(len(header))] + print(' '.join(h.ljust(w) for h, w in zip(header, widths))) + for func, values in results: + row = [func.name, f'{func.start:#x}'] + [f'{values[m]:.2f}' for m in METRICS] + print(' '.join(c.ljust(w) for c, w in zip(row, widths))) + + if args.csv: + with open(args.csv, 'w', newline='') as f: + writer = csv.writer(f) + writer.writerow(['function', 'address'] + METRICS) + for func, values in results: + writer.writerow([func.name, hex(func.start)] + [values[m] for m in METRICS]) + binaryninja.log_info(f'log_function_complexity: wrote {len(results)} row(s) to {args.csv}') + + return 0 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/python/function.py b/python/function.py index 1dfaf9b691..77b0c01cbb 100644 --- a/python/function.py +++ b/python/function.py @@ -38,6 +38,7 @@ from . import lowlevelil from . import mediumlevelil from . import highlevelil +from . import complexity from . import binaryview from . import basicblock from . import databuffer @@ -1182,6 +1183,36 @@ def hlil_if_available(self) -> Optional['highlevelil.HighLevelILFunction']: return None return highlevelil.HighLevelILFunction(self.arch, result, self) + def get_complexity(self, metric: str = 'composite') -> float: + """ + Compute a code complexity score for this function. + + Several independent metrics are available, each isolating a different factor (code + length, number/density of branches, diversity of instruction types, nesting depth of + control flow, or a weighted blend of all of the above). See + :py:mod:`binaryninja.complexity` for a description of each one. + + :param metric: Name of the metric to compute. See :py:meth:`complexity_metrics` for the + list of valid names. Defaults to ``"composite"``. + :return: The computed complexity score. Higher means more complex; scales differ between + metrics, so only compare scores computed with the same metric. + :raises ValueError: If `metric` is not a recognized metric name. + + :Example: + >>> current_function.get_complexity("cyclomatic") + 3.0 + >>> current_function.get_complexity("composite") + 7.25 + """ + return complexity.get_code_complexity(self, metric) + + @staticmethod + def complexity_metrics() -> List[str]: + """ + Names accepted by :py:meth:`get_complexity`. + """ + return complexity.list_complexity_metrics() + @property def pseudo_c(self) -> Optional['languagerepresentation.LanguageRepresentationFunction']: return self.language_representation("Pseudo C") diff --git a/rust/src/function.rs b/rust/src/function.rs index 5023b958f2..a1c9310318 100644 --- a/rust/src/function.rs +++ b/rust/src/function.rs @@ -27,7 +27,7 @@ use crate::{ platform::Platform, references::CodeReference, string::*, - symbol::{Binding, Symbol}, + symbol::{Binding, Symbol, SymbolType}, tags::{Tag, TagReference, TagType}, types::{IntegerDisplayType, QualifiedName, ReturnValue, Type, ValueLocation}, }; @@ -45,14 +45,16 @@ use crate::confidence::Conf; use crate::high_level_il::HighLevelILFunction; use crate::language_representation::CoreLanguageRepresentationFunction; use crate::low_level_il::LowLevelILRegularFunction; -use crate::medium_level_il::MediumLevelILFunction; +use crate::medium_level_il::{ + MediumLevelILFunction, MediumLevelILLiftedInstruction, MediumLevelILLiftedInstructionKind, +}; use crate::metadata::Metadata; use crate::variable::{ MergedVariable, NamedVariableWithType, RegisterValue, RegisterValueType, StackVariableReference, Variable, }; use crate::workflow::Workflow; -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use std::ffi::CStr; use std::fmt::{Debug, Formatter}; use std::ptr::NonNull; @@ -654,6 +656,101 @@ impl Function { } } + // Code complexity primitives - thin wrappers over the matching BNGetFunction*Complexity Core + // calls, each a single whole-function walk. + pub fn cyclomatic_complexity(&self) -> f64 { + unsafe { BNGetFunctionCyclomaticComplexity(self.handle) } + } + + pub fn instruction_count_complexity(&self) -> f64 { + unsafe { BNGetFunctionInstructionCountComplexity(self.handle) } + } + + pub fn token_count_complexity(&self) -> f64 { + unsafe { BNGetFunctionTokenCountComplexity(self.handle) } + } + + pub fn branch_density_complexity(&self) -> f64 { + unsafe { BNGetFunctionBranchDensityComplexity(self.handle) } + } + + pub fn halstead_complexity(&self) -> f64 { + unsafe { BNGetFunctionHalsteadComplexity(self.handle) } + } + + pub fn nesting_depth_complexity(&self) -> f64 { + unsafe { BNGetFunctionNestingDepthComplexity(self.handle) } + } + + pub fn cognitive_complexity(&self) -> f64 { + unsafe { BNGetFunctionCognitiveComplexity(self.handle) } + } + + /// A single weighted score blending the individual primitives above. Weights are tuned so + /// that a small, simple function scores in the single digits and a deeply nested, branchy + /// function scores in the hundreds. Callers who want different behavior should call the + /// primitives directly and combine them with their own weights. + /// + /// Implemented here rather than in Core: it's only a bounded number of whole-function calls, + /// not a per-instruction walk, so there's nothing to gain from Core doing the aggregation. + pub fn composite_complexity(&self) -> f64 { + const CYCLOMATIC_WEIGHT: f64 = 0.25; + const INSTRUCTION_COUNT_WEIGHT: f64 = 0.05; + const BRANCH_DENSITY_WEIGHT: f64 = 10.0; + const HALSTEAD_WEIGHT: f64 = 0.05; + const NESTING_DEPTH_WEIGHT: f64 = 2.0; + const COGNITIVE_WEIGHT: f64 = 0.5; + + CYCLOMATIC_WEIGHT * self.cyclomatic_complexity() + + INSTRUCTION_COUNT_WEIGHT * self.instruction_count_complexity() + + BRANCH_DENSITY_WEIGHT * self.branch_density_complexity() + + HALSTEAD_WEIGHT * self.halstead_complexity() + + NESTING_DEPTH_WEIGHT * self.nesting_depth_complexity() + + COGNITIVE_WEIGHT * self.cognitive_complexity() + } + + /// How much, and how riskily, this function fans out into other code - captures the "main() + /// is just a dispatcher" shape that every intraprocedural metric misses. Distinct callees + /// count in full; repeated calls to a callee already counted (e.g. in a loop) count for less. + /// Calls to external/library code count more, since we can't inspect what they do. Indirect + /// calls count the most, since we can't even statically know what runs. + pub fn fan_out_complexity(&self) -> f64 { + let calls = classify_call_sites(self); + + let mut distinct_targets = HashSet::new(); + for callee in &calls.internal_callees { + distinct_targets.insert(callee.start()); + } + + let repeat_calls = calls.internal_callees.len() - distinct_targets.len(); + + distinct_targets.len() as f64 + + 0.25 * repeat_calls as f64 + + 1.5 * calls.external_count as f64 + + 2.5 * calls.indirect_count as f64 + } + + /// How many places in this binary's code reference this function - the complement of + /// [`Function::fan_out_complexity`]: that measures how much this function reaches into other + /// code, this measures how much other code reaches into it. Not "complexity" in the same + /// sense as the other metrics - a heavily-referenced function isn't necessarily hard to read + /// on its own - but it's a direct measure of blast radius. + pub fn code_reference_complexity(&self) -> f64 { + self.view().code_refs_to_addr(self.start()).len() as f64 + } + + /// This function's own composite complexity plus the (decayed) composite complexity of + /// everything it calls, recursively, up to two hops. This is the metric that actually answers + /// "this function looks simple, but how much do I have to read to understand what it does": a + /// function with a low `composite` score but a high `transitive` score is exactly a thin + /// dispatcher over a lot of real work. + pub fn transitive_complexity(&self) -> f64 { + const MAX_DEPTH: usize = 2; + const DECAY: f64 = 0.5; + let mut visited = HashSet::new(); + transitive_complexity_walk(self, MAX_DEPTH, DECAY, &mut visited) + } + pub fn return_type(&self) -> Conf> { let raw_return_type = unsafe { BNGetFunctionReturnType(self.handle) }; Conf::>::from_owned_raw(raw_return_type) @@ -2753,6 +2850,125 @@ impl Function { } } +// One entry per call site that resolves to a function defined in this binary (duplicates +// included, e.g. the same helper called twice shows up twice). Backs fan_out_complexity and +// transitive_complexity. +#[derive(Default)] +struct CallSites { + internal_callees: Vec>, + external_count: usize, + indirect_count: usize, +} + +// A call-like instruction's dest operand resolves to a target address only when it's a plain +// constant/pointer/import - anything else (a register, a loaded value, ...) is an indirect call +// we can't statically chase. +fn resolve_call_dest(dest: &MediumLevelILLiftedInstruction) -> Option { + use MediumLevelILLiftedInstructionKind::*; + match &dest.kind { + Const(c) | ConstPtr(c) | Import(c) => Some(c.constant), + _ => None, + } +} + +// Walks every call-like MLIL statement in `func` and sorts it into one of three buckets: +// +// * resolves to a function defined in this binary -> appended to internal_callees +// * resolves to an address, but that address is an import/thunk/library stub, or analysis +// couldn't identify a function there -> external_count +// * the target isn't a static constant at all (computed/register call, e.g. through a function +// pointer or vtable) -> indirect_count, since we can't know what runs there without deeper (and +// possibly incomplete) analysis +// +// Built entirely from already-public API calls (MLIL iteration, BinaryView lookups) - this used +// to be Core-internal, but nothing here needs a per-instruction callback into Core, so there's +// nothing to gain from Core doing it on our behalf. +fn classify_call_sites(func: &Function) -> CallSites { + use MediumLevelILLiftedInstructionKind::*; + + let mut result = CallSites::default(); + + let Ok(mlil) = func.medium_level_il() else { + return result; + }; + + let view = func.view(); + + for block in mlil.basic_blocks().iter() { + for instr in block.iter() { + let lifted = instr.lift(); + + let dest = match &lifted.kind { + Syscall(_) | SyscallUntyped(_) => { + result.external_count += 1; + continue; + } + Call(op) | Tailcall(op) => op.dest.as_ref(), + CallUntyped(op) | TailcallUntyped(op) => op.dest.as_ref(), + _ => continue, + }; + + let Some(target_addr) = resolve_call_dest(dest) else { + result.indirect_count += 1; + continue; + }; + + let targets = view.functions_at(target_addr); + if targets.is_empty() { + result.external_count += 1; + continue; + } + + let is_external = view.symbol_by_address(target_addr).is_some_and(|symbol| { + matches!( + symbol.sym_type(), + SymbolType::ImportAddress + | SymbolType::ImportedFunction + | SymbolType::External + | SymbolType::LibraryFunction + ) + }); + + if is_external { + result.external_count += 1; + } else { + result.internal_callees.push(targets.get(0).clone()); + } + } + } + + result +} + +// `func`'s own composite complexity plus the (decayed) composite complexity of everything it +// calls, recursively, up to `depth` hops. Each function is only ever counted once (tracked by +// start address) across the whole walk, which both breaks recursion/cycles safely and avoids +// over-counting a shared helper reached through multiple paths. +fn transitive_complexity_walk( + func: &Function, + depth: usize, + decay: f64, + visited: &mut HashSet, +) -> f64 { + if !visited.insert(func.start()) { + return 0.0; + } + + let total = func.composite_complexity(); + if depth == 0 { + return total; + } + + let mut callees = HashMap::new(); + for callee in classify_call_sites(func).internal_callees { + callees.entry(callee.start()).or_insert(callee); + } + + callees.values().fold(total, |acc, callee| { + acc + decay * transitive_complexity_walk(callee, depth - 1, decay, visited) + }) +} + impl Debug for Function { fn fmt(&self, f: &mut Formatter) -> std::fmt::Result { // TODO: I am sure there is more we should add to this.