callgraph

package
v0.0.0-...-bfec689 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Oct 31, 2025 License: AGPL-3.0 Imports: 9 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func ExtractVariableAssignments

func ExtractVariableAssignments(
	filePath string,
	sourceCode []byte,
	typeEngine *TypeInferenceEngine,
	registry *ModuleRegistry,
	builtinRegistry *BuiltinRegistry,
) error

ExtractVariableAssignments extracts variable assignments from a Python file and populates the type inference engine with inferred types.

Algorithm:

  1. Parse source code with tree-sitter Python parser
  2. Traverse AST to find assignment statements
  3. For each assignment: - Extract variable name - Infer type from RHS (literal, function call, or method call) - Create VariableBinding with inferred type - Add binding to function scope

Parameters:

  • filePath: absolute path to the Python file
  • sourceCode: contents of the file as byte array
  • typeEngine: type inference engine to populate
  • registry: module registry for resolving module paths
  • builtinRegistry: builtin types registry for literal inference

Returns:

  • error: if parsing fails

func GetFrameworkCategory

func GetFrameworkCategory(fqn string) string

GetFrameworkCategory returns the category of a framework given its FQN. Returns empty string if not a known framework.

func GetFrameworkName

func GetFrameworkName(fqn string) string

GetFrameworkName returns the name of a framework given its FQN. Returns empty string if not a known framework.

func InitializeCallGraph

func InitializeCallGraph(codeGraph *graph.CodeGraph, projectRoot string) (*CallGraph, *ModuleRegistry, *PatternRegistry, error)

InitializeCallGraph builds the call graph from a code graph. This integrates the 3-pass algorithm into the main initialization pipeline.

Algorithm:

  1. Build module registry from project directory
  2. Build call graph from code graph using registry
  3. Load default security patterns
  4. Return integrated result

Parameters:

  • codeGraph: the parsed code graph from Initialize()
  • projectRoot: absolute path to project root directory

Returns:

  • CallGraph: complete call graph with edges and call sites
  • ModuleRegistry: module path mappings
  • PatternRegistry: loaded security patterns
  • error: if any step fails

func IsDjangoORMPattern

func IsDjangoORMPattern(target string) (bool, string)

IsDjangoORMPattern checks if a call target matches Django ORM pattern. Django ORM pattern: ModelName.objects.<method>

Examples:

  • "Task.objects.filter" → true
  • "User.objects.get" → true
  • "Annotation.objects.all" → true
  • "task.save" → false (instance method, not manager)

Parameters:

  • target: call target string (e.g., "Task.objects.filter")

Returns:

  • true if it matches Django ORM pattern
  • the method name if matched (e.g., "filter")

func IsORMPattern

func IsORMPattern(target string) (bool, string, string)

IsORMPattern checks if a call target matches any known ORM pattern.

Parameters:

  • target: call target string

Returns:

  • true if it matches any ORM pattern
  • the ORM pattern name (e.g., "Django ORM")
  • the method name (e.g., "filter")

func IsSQLAlchemyORMPattern

func IsSQLAlchemyORMPattern(target string) (bool, string)

IsSQLAlchemyORMPattern checks if a call target matches SQLAlchemy ORM pattern. SQLAlchemy patterns are more varied, but common ones include:

  • session.query(Model).filter(...)
  • db.session.query(Model).all()
  • Model.query.filter_by(...)

Parameters:

  • target: call target string

Returns:

  • true if it matches SQLAlchemy ORM pattern
  • the method name if matched

func MergeReturnTypes

func MergeReturnTypes(statements []*ReturnStatement) map[string]*TypeInfo

MergeReturnTypes combines multiple return statements for same function. Takes the highest confidence return type.

func ResolveDjangoORMCall

func ResolveDjangoORMCall(target string, modulePath string, registry *ModuleRegistry, codeGraph *graph.CodeGraph) (string, bool)

ResolveDjangoORMCall attempts to resolve a Django ORM call pattern. It constructs a synthetic FQN for the ORM method even though it doesn't exist in source code, because Django generates these methods at runtime.

Parameters:

  • target: the call target (e.g., "Task.objects.filter")
  • modulePath: the current module path
  • registry: module registry
  • codeGraph: the parsed code graph (for model validation)

Returns:

  • fully qualified name for the ORM call
  • true if successfully resolved as Django ORM

func ResolveORMCall

func ResolveORMCall(target string, modulePath string, registry *ModuleRegistry, codeGraph *graph.CodeGraph) (string, bool)

ResolveORMCall attempts to resolve any ORM pattern.

Parameters:

  • target: the call target
  • modulePath: the current module path
  • registry: module registry
  • codeGraph: the parsed code graph

Returns:

  • fully qualified name for the ORM call
  • true if successfully resolved as any ORM pattern

func ResolveSQLAlchemyORMCall

func ResolveSQLAlchemyORMCall(target string, modulePath string) (string, bool)

ResolveSQLAlchemyORMCall attempts to resolve a SQLAlchemy ORM call pattern.

Parameters:

  • target: the call target
  • modulePath: the current module path

Returns:

  • fully qualified name for the ORM call
  • true if successfully resolved as SQLAlchemy ORM

func ValidateDjangoModel

func ValidateDjangoModel(modelName string, codeGraph *graph.CodeGraph) bool

ValidateDjangoModel checks if a name is likely a Django model by examining the code graph for the class definition and checking if it inherits from django.db.models.Model or has "Model" in its name.

This is a heuristic check since we can't always definitively determine if something is a Django model without runtime information.

Parameters:

  • modelName: the name to check (e.g., "Task", "User")
  • codeGraph: the parsed code graph

Returns:

  • true if the name is likely a Django model

Types

type Argument

type Argument struct {
	Value      string // The argument expression as a string
	IsVariable bool   // Whether this argument is a variable reference
	Position   int    // Position in the argument list (0-indexed)
}

Argument represents a single argument passed to a function call. Tracks both the value/expression and metadata about the argument.

type BasicBlock

type BasicBlock struct {
	// ID uniquely identifies this block within the CFG
	ID string

	// Type categorizes the block for analysis purposes
	Type BlockType

	// StartLine is the first line of code in this block (1-indexed)
	StartLine int

	// EndLine is the last line of code in this block (1-indexed)
	EndLine int

	// Instructions contains the call sites within this block.
	// Call sites represent function/method invocations that occur
	// during execution of this block.
	Instructions []CallSite

	// Successors are the blocks that can execute after this block.
	// For normal blocks: single successor
	// For conditional blocks: two successors (true/false branches)
	// For switch blocks: multiple successors (one per case)
	// For exit blocks: empty (no successors)
	Successors []string

	// Predecessors are the blocks that can execute before this block.
	// Used for backward analysis and dominance calculations.
	Predecessors []string

	// Condition stores the condition expression for conditional blocks.
	// Empty for non-conditional blocks.
	// Examples: "x > 0", "user.is_admin()", "data is not None"
	Condition string

	// Dominators are the blocks that always execute before this block
	// on any path from entry. Used for security analysis to determine
	// if sanitization always occurs before usage.
	Dominators []string
}

BasicBlock represents a basic block in a control flow graph. A basic block is a maximal sequence of instructions with:

  • Single entry point (at the beginning)
  • Single exit point (at the end)
  • No internal branches

Basic blocks are the nodes in a CFG, connected by edges representing control flow between blocks.

type BlockType

type BlockType string

BlockType represents the type of basic block in a control flow graph. Different block types enable different security analysis patterns.

const (
	// BlockTypeEntry represents the entry point of a function.
	// Every function has exactly one entry block.
	BlockTypeEntry BlockType = "entry"

	// BlockTypeExit represents the exit point of a function.
	// Every function has exactly one exit block where all return paths converge.
	BlockTypeExit BlockType = "exit"

	// BlockTypeNormal represents a regular basic block with sequential execution.
	// Contains straight-line code with no branches.
	BlockTypeNormal BlockType = "normal"

	// BlockTypeConditional represents a conditional branch block.
	// Has multiple successor blocks (true/false branches).
	// Examples: if statements, ternary operators, short-circuit logic.
	BlockTypeConditional BlockType = "conditional"

	// BlockTypeLoop represents a loop header block.
	// Has back-edges for loop iteration.
	// Examples: while loops, for loops, do-while loops.
	BlockTypeLoop BlockType = "loop"

	// BlockTypeSwitch represents a switch/match statement block.
	// Has multiple successor blocks (one per case).
	BlockTypeSwitch BlockType = "switch"

	// BlockTypeTry represents a try block in exception handling.
	// Has normal successor and exception handler successors.
	BlockTypeTry BlockType = "try"

	// BlockTypeCatch represents a catch/except block in exception handling.
	// Handles exceptions from try blocks.
	BlockTypeCatch BlockType = "catch"

	// BlockTypeFinally represents a finally block in exception handling.
	// Always executes regardless of exceptions.
	BlockTypeFinally BlockType = "finally"
)

type BuiltinMethod

type BuiltinMethod struct {
	Name       string    // Method name (e.g., "upper", "append")
	ReturnType *TypeInfo // Return type of the method
}

BuiltinMethod represents a method available on a builtin type.

type BuiltinRegistry

type BuiltinRegistry struct {
	Types map[string]*BuiltinType // Type FQN -> builtin type info
}

BuiltinRegistry maintains information about Python builtin types and their methods. This enables type inference for literal values and builtin method calls.

func NewBuiltinRegistry

func NewBuiltinRegistry() *BuiltinRegistry

NewBuiltinRegistry creates and initializes a registry with Python builtin types. The registry is pre-populated with common types: str, list, dict, set, tuple, int, float, bool, bytes, and their associated methods.

Returns:

  • Initialized BuiltinRegistry with all builtin types

func (*BuiltinRegistry) GetMethod

func (br *BuiltinRegistry) GetMethod(typeFQN, methodName string) *BuiltinMethod

GetMethod retrieves method information for a builtin type.

Parameters:

  • typeFQN: fully qualified type name
  • methodName: name of the method

Returns:

  • BuiltinMethod if found, nil otherwise

func (*BuiltinRegistry) GetType

func (br *BuiltinRegistry) GetType(typeFQN string) *BuiltinType

GetType retrieves builtin type information by its fully qualified name.

Parameters:

  • typeFQN: fully qualified type name (e.g., "builtins.str")

Returns:

  • BuiltinType if found, nil otherwise

func (*BuiltinRegistry) InferLiteralType

func (br *BuiltinRegistry) InferLiteralType(literal string) *TypeInfo

InferLiteralType infers the type of a Python literal value. Supports: strings, integers, floats, booleans, lists, dicts, sets, tuples.

Parameters:

  • literal: the literal value as a string

Returns:

  • TypeInfo with confidence 1.0 if recognized, nil otherwise

type BuiltinType

type BuiltinType struct {
	FQN     string                    // Fully qualified name (e.g., "builtins.str")
	Methods map[string]*BuiltinMethod // Method name -> method info
}

BuiltinType represents a Python builtin type with its available methods.

type CallGraph

type CallGraph struct {
	// Forward edges: maps fully qualified function name to list of functions it calls
	// Key: caller FQN (e.g., "myapp.views.get_user")
	// Value: list of callee FQNs (e.g., ["myapp.db.query", "myapp.utils.sanitize"])
	Edges map[string][]string

	// Reverse edges: maps fully qualified function name to list of functions that call it
	// Useful for backward slicing and finding all callers of a function
	// Key: callee FQN
	// Value: list of caller FQNs
	ReverseEdges map[string][]string

	// Detailed call site information for each function
	// Key: caller FQN
	// Value: list of all call sites within that function
	CallSites map[string][]CallSite

	// Map from fully qualified name to the actual function node in the graph
	// This allows quick lookup of function metadata (line number, file, etc.)
	Functions map[string]*graph.Node
}

CallGraph represents the complete call graph of a program. It maps function definitions to their call sites and provides both forward (callers → callees) and reverse (callees → callers) edges.

Example:

Function A calls B and C
edges: {"A": ["B", "C"]}
reverseEdges: {"B": ["A"], "C": ["A"]}

func BuildCallGraph

func BuildCallGraph(codeGraph *graph.CodeGraph, registry *ModuleRegistry, projectRoot string) (*CallGraph, error)

BuildCallGraph constructs the complete call graph for a Python project. This is Pass 3 of the 3-pass algorithm:

  • Pass 1: BuildModuleRegistry - map files to modules
  • Pass 2: ExtractImports + ExtractCallSites - parse imports and calls
  • Pass 3: BuildCallGraph - resolve calls and build graph

Algorithm:

  1. For each Python file in the project: a. Extract imports to build ImportMap b. Extract call sites from AST c. Extract function definitions from main graph
  2. For each call site: a. Resolve target name using ImportMap b. Find target function definition in registry c. Add edge from caller to callee d. Store detailed call site information

Parameters:

  • codeGraph: the existing code graph with parsed AST nodes
  • registry: module registry mapping files to modules
  • projectRoot: absolute path to project root

Returns:

  • CallGraph: complete call graph with edges and call sites
  • error: if any step fails

Example:

Given:
  File: myapp/views.py
    def get_user():
        sanitize(data)  # call to myapp.utils.sanitize

Creates:
  edges: {"myapp.views.get_user": ["myapp.utils.sanitize"]}
  reverseEdges: {"myapp.utils.sanitize": ["myapp.views.get_user"]}
  callSites: {"myapp.views.get_user": [CallSite{Target: "sanitize", ...}]}

func NewCallGraph

func NewCallGraph() *CallGraph

NewCallGraph creates and initializes a new CallGraph instance. All maps are pre-allocated to avoid nil pointer issues.

func (*CallGraph) AddCallSite

func (cg *CallGraph) AddCallSite(caller string, callSite CallSite)

AddCallSite adds a call site to the call graph. This stores detailed information about where and how a function is called.

Parameters:

  • caller: fully qualified name of the calling function
  • callSite: detailed information about the call

func (*CallGraph) AddEdge

func (cg *CallGraph) AddEdge(caller, callee string)

AddEdge adds a directed edge from caller to callee in the call graph. Automatically updates both forward and reverse edges.

Parameters:

  • caller: fully qualified name of the calling function
  • callee: fully qualified name of the called function

func (*CallGraph) GetCallees

func (cg *CallGraph) GetCallees(caller string) []string

GetCallees returns all functions called by the specified function. Uses the forward edges for efficient lookup.

Parameters:

  • caller: fully qualified name of the function

Returns:

  • list of callee FQNs, or empty slice if no callees found

func (*CallGraph) GetCallers

func (cg *CallGraph) GetCallers(callee string) []string

GetCallers returns all functions that call the specified function. Uses the reverse edges for efficient lookup.

Parameters:

  • callee: fully qualified name of the function

Returns:

  • list of caller FQNs, or empty slice if no callers found

type CallSite

type CallSite struct {
	Target        string     // The name of the function being called (e.g., "eval", "utils.sanitize")
	Location      Location   // Where this call occurs in the source code
	Arguments     []Argument // Arguments passed to the call
	Resolved      bool       // Whether we successfully resolved this call to a definition
	TargetFQN     string     // Fully qualified name after resolution (e.g., "myapp.utils.sanitize")
	FailureReason string     // Why resolution failed (empty if Resolved=true)

	// Phase 2: Type inference metadata
	ResolvedViaTypeInference bool    // Was this resolved using type inference?
	InferredType             string  // The inferred type FQN (e.g., "builtins.str", "test.User")
	TypeConfidence           float32 // Confidence score of the type inference (0.0-1.0)
	TypeSource               string  // How type was inferred (e.g., "literal", "return_type", "class_instantiation")
}

CallSite represents a function/method call location in the source code. It captures both the syntactic information (where the call is) and semantic information (what is being called and with what arguments).

func ExtractCallSites

func ExtractCallSites(filePath string, sourceCode []byte, importMap *ImportMap) ([]*CallSite, error)

ExtractCallSites extracts all function/method call sites from a Python file. It traverses the AST to find call expressions and builds CallSite objects with caller context, callee information, and arguments.

Algorithm:

  1. Parse source code with tree-sitter Python parser
  2. Traverse AST to find call expressions
  3. For each call, extract: - Caller function/method (containing context) - Callee name (function/method being called) - Arguments (positional and keyword) - Source location (file, line, column)
  4. Build CallSite objects for each call

Parameters:

  • filePath: absolute path to the Python file being analyzed
  • sourceCode: contents of the Python file as byte array
  • importMap: import mappings for resolving qualified names

Returns:

  • []CallSite: list of all call sites found in the file
  • error: if parsing fails or source is invalid

Example:

Source code:
  def process_data():
      result = sanitize(data)
      db.query(result)

Extracts CallSites:
  [
    {Caller: "process_data", Callee: "sanitize", Args: ["data"]},
    {Caller: "process_data", Callee: "db.query", Args: ["result"]}
  ]

type ControlFlowGraph

type ControlFlowGraph struct {
	// FunctionFQN is the fully qualified name of the function this CFG represents
	FunctionFQN string

	// Blocks maps block IDs to BasicBlock objects
	Blocks map[string]*BasicBlock

	// EntryBlockID identifies the entry block
	EntryBlockID string

	// ExitBlockID identifies the exit block
	ExitBlockID string

	// CallGraph reference for resolving inter-procedural flows
	CallGraph *CallGraph
}

ControlFlowGraph represents the control flow graph of a function. A CFG models all possible execution paths through a function, enabling data flow and taint analysis for security vulnerabilities.

Example:

def process_user(user_id):
    user = get_user(user_id)        # Block 1 (entry)
    if user.is_admin():              # Block 2 (conditional)
        grant_access()               # Block 3 (true branch)
    else:
        deny_access()                # Block 4 (false branch)
    log_action(user)                 # Block 5 (merge point)
    return                           # Block 6 (exit)

CFG Structure:

Entry → Block1 → Block2 → Block3 → Block5 → Exit
                       → Block4 ↗

func NewControlFlowGraph

func NewControlFlowGraph(functionFQN string) *ControlFlowGraph

NewControlFlowGraph creates and initializes a new CFG for a function.

func (*ControlFlowGraph) AddBlock

func (cfg *ControlFlowGraph) AddBlock(block *BasicBlock)

AddBlock adds a basic block to the CFG.

func (*ControlFlowGraph) AddEdge

func (cfg *ControlFlowGraph) AddEdge(fromBlockID, toBlockID string)

AddEdge adds a control flow edge from one block to another. Automatically updates both successors and predecessors.

func (*ControlFlowGraph) ComputeDominators

func (cfg *ControlFlowGraph) ComputeDominators()

ComputeDominators calculates dominator sets for all blocks. A block X dominates block Y if every path from entry to Y must go through X. This is essential for determining if sanitization always occurs before usage.

Algorithm: Iterative data flow analysis

  1. Initialize: Entry dominates only itself, all others dominated by all blocks
  2. Iterate until fixed point: For each block B (except entry): Dom(B) = {B} ∪ (intersection of Dom(P) for all predecessors P of B)

func (*ControlFlowGraph) GetAllPaths

func (cfg *ControlFlowGraph) GetAllPaths() [][]string

GetAllPaths returns all execution paths from entry to exit. Used for exhaustive security analysis. WARNING: Can be exponential in size for complex CFGs with loops.

func (*ControlFlowGraph) GetBlock

func (cfg *ControlFlowGraph) GetBlock(blockID string) (*BasicBlock, bool)

GetBlock retrieves a block by ID.

func (*ControlFlowGraph) GetPredecessors

func (cfg *ControlFlowGraph) GetPredecessors(blockID string) []*BasicBlock

GetPredecessors returns the predecessor blocks of a given block.

func (*ControlFlowGraph) GetSuccessors

func (cfg *ControlFlowGraph) GetSuccessors(blockID string) []*BasicBlock

GetSuccessors returns the successor blocks of a given block.

func (*ControlFlowGraph) IsDominator

func (cfg *ControlFlowGraph) IsDominator(dominator, dominated string) bool

IsDominator returns true if dominator dominates dominated. Used to check if sanitization (in dominator) always occurs before usage (in dominated).

type FrameworkDefinition

type FrameworkDefinition struct {
	Name        string   // Display name (e.g., "Django")
	Prefixes    []string // Module prefixes to match (e.g., ["django.", "django"])
	Description string   // Human-readable description
	Category    string   // Category: "web", "orm", "testing", "stdlib", etc.
}

FrameworkDefinition represents a known external framework or library. This is used to mark calls to external code as resolved, even though we don't have the source code for these frameworks.

func IsKnownFramework

func IsKnownFramework(fqn string) (bool, *FrameworkDefinition)

IsKnownFramework checks if the given fully qualified name (FQN) belongs to a known external framework or standard library.

Parameters:

  • fqn: fully qualified name (e.g., "django.db.models.ForeignKey")

Returns:

  • true if the FQN matches any known framework
  • the matching framework definition

func LoadFrameworks

func LoadFrameworks() []FrameworkDefinition

LoadFrameworks returns the list of known frameworks. This function provides an extensibility hook for future enhancements where frameworks might be loaded from a configuration file.

type FunctionScope

type FunctionScope struct {
	FunctionFQN string                      // Fully qualified name of the function
	Variables   map[string]*VariableBinding // Variable name -> binding
	ReturnType  *TypeInfo                   // Inferred return type of the function
}

FunctionScope represents the type environment within a function. It tracks variable types and return type for a specific function.

func NewFunctionScope

func NewFunctionScope(functionFQN string) *FunctionScope

NewFunctionScope creates a new function scope with initialized maps.

Parameters:

  • functionFQN: fully qualified name of the function

Returns:

  • Initialized FunctionScope

type ImportMap

type ImportMap struct {
	FilePath string            // Absolute path to the file containing these imports
	Imports  map[string]string // Maps alias/name to fully qualified module path
}

ImportMap represents the import statements in a single Python file. Maps local aliases to fully qualified module paths.

Example:

File contains: from myapp.utils import sanitize as clean
Imports: {"clean": "myapp.utils.sanitize"}

func ExtractImports

func ExtractImports(filePath string, sourceCode []byte, registry *ModuleRegistry) (*ImportMap, error)

ExtractImports extracts all import statements from a Python file and builds an ImportMap. It handles four main import styles:

  1. Simple imports: import module
  2. From imports: from module import name
  3. Aliased imports: from module import name as alias
  4. Relative imports: from . import module, from .. import module

The resulting ImportMap maps local names (aliases or imported names) to their fully qualified module paths, enabling later resolution of function calls.

Algorithm:

  1. Parse source code with tree-sitter Python parser
  2. Traverse AST to find all import statements
  3. Process each import to extract module paths and aliases
  4. Resolve relative imports using module registry
  5. Build ImportMap with resolved fully qualified names

Parameters:

  • filePath: absolute path to the Python file being analyzed
  • sourceCode: contents of the Python file as byte array
  • registry: module registry for resolving module paths and relative imports

Returns:

  • ImportMap: map of local names to fully qualified module paths
  • error: if parsing fails or source is invalid

Example:

Source code:
  import os
  from myapp.utils import sanitize
  from myapp.db import query as db_query
  from . import helper
  from ..config import settings

Result ImportMap:
  {
    "os": "os",
    "sanitize": "myapp.utils.sanitize",
    "db_query": "myapp.db.query",
    "helper": "myapp.submodule.helper",
    "settings": "myapp.config.settings"
  }

func NewImportMap

func NewImportMap(filePath string) *ImportMap

NewImportMap creates and initializes a new ImportMap instance.

func (*ImportMap) AddImport

func (im *ImportMap) AddImport(alias, fqn string)

AddImport adds an import mapping to the import map.

Parameters:

  • alias: the local name used in the file (e.g., "clean", "sanitize", "utils")
  • fqn: the fully qualified name (e.g., "myapp.utils.sanitize")

func (*ImportMap) Resolve

func (im *ImportMap) Resolve(alias string) (string, bool)

Resolve looks up the fully qualified name for a local alias.

Parameters:

  • alias: the local name to resolve

Returns:

  • fully qualified name and true if found, empty string and false otherwise

type ImportMapCache

type ImportMapCache struct {
	// contains filtered or unexported fields
}

ImportMapCache provides thread-safe caching of ImportMap instances. This avoids re-parsing imports from the same file multiple times.

The cache uses a read-write mutex to allow concurrent reads while ensuring safe writes. This is critical for performance since:

  • Import extraction involves tree-sitter parsing (expensive)
  • Many files may import the same modules
  • Build call graph processes files sequentially (for now)

Example usage:

cache := NewImportMapCache()
importMap := cache.GetOrExtract(filePath, sourceCode, registry)

func NewImportMapCache

func NewImportMapCache() *ImportMapCache

NewImportMapCache creates a new empty import map cache.

func (*ImportMapCache) Get

func (c *ImportMapCache) Get(filePath string) (*ImportMap, bool)

Get retrieves an ImportMap from the cache if it exists.

Parameters:

  • filePath: absolute path to the Python file

Returns:

  • ImportMap and true if found in cache, nil and false otherwise

func (*ImportMapCache) GetOrExtract

func (c *ImportMapCache) GetOrExtract(filePath string, sourceCode []byte, registry *ModuleRegistry) (*ImportMap, error)

GetOrExtract retrieves an ImportMap from cache or extracts it if not cached. This is the main entry point for using the cache.

Parameters:

  • filePath: absolute path to the Python file
  • sourceCode: file contents (only used if extraction needed)
  • registry: module registry for resolving imports

Returns:

  • ImportMap from cache or newly extracted
  • error if extraction fails (cache misses only)

Thread-safety:

  • Multiple goroutines can safely call GetOrExtract concurrently
  • First caller for a file will extract and cache
  • Subsequent callers will get cached result

func (*ImportMapCache) Put

func (c *ImportMapCache) Put(filePath string, importMap *ImportMap)

Put stores an ImportMap in the cache.

Parameters:

  • filePath: absolute path to the Python file
  • importMap: the extracted ImportMap to cache

type Location

type Location struct {
	File   string // Absolute path to the source file
	Line   int    // Line number (1-indexed)
	Column int    // Column number (1-indexed)
}

Location represents a source code location for tracking call sites. This enables precise mapping of where calls occur in the source code.

type ModuleRegistry

type ModuleRegistry struct {
	// Maps fully qualified module path to absolute file path
	// Key: "myapp.utils.helpers"
	// Value: "/absolute/path/to/myapp/utils/helpers.py"
	Modules map[string]string

	// Maps absolute file path to fully qualified module path (reverse of Modules)
	// Key: "/absolute/path/to/myapp/utils/helpers.py"
	// Value: "myapp.utils.helpers"
	// Used for resolving relative imports
	FileToModule map[string]string

	// Maps short module names to all matching file paths (handles ambiguity)
	// Key: "helpers"
	// Value: ["/path/to/myapp/utils/helpers.py", "/path/to/lib/helpers.py"]
	ShortNames map[string][]string

	// Cache for resolved imports to avoid redundant lookups
	// Key: import string (e.g., "utils.helpers")
	// Value: fully qualified module path
	ResolvedImports map[string]string
}

ModuleRegistry maintains the mapping between Python file paths and module paths. This is essential for resolving imports and building fully qualified names.

Example:

File: /project/myapp/utils/helpers.py
Module: myapp.utils.helpers

func BuildModuleRegistry

func BuildModuleRegistry(rootPath string) (*ModuleRegistry, error)

BuildModuleRegistry walks a directory tree and builds a complete module registry. It discovers all Python files and maps them to their corresponding module paths.

The registry enables:

  • Resolving fully qualified names (FQNs) for functions
  • Mapping import statements to actual files
  • Detecting ambiguous module names

Algorithm:

  1. Walk directory tree recursively
  2. Skip common non-source directories (venv, __pycache__, etc.)
  3. Convert file paths to Python module paths
  4. Index both full module paths and short names

Parameters:

  • rootPath: absolute path to the project root directory

Returns:

  • ModuleRegistry: populated registry with all discovered modules
  • error: if root path doesn't exist or is inaccessible

Example:

registry, err := BuildModuleRegistry("/path/to/myapp")
// Discovers:
//   /path/to/myapp/views.py → "myapp.views"
//   /path/to/myapp/utils/helpers.py → "myapp.utils.helpers"

func NewModuleRegistry

func NewModuleRegistry() *ModuleRegistry

NewModuleRegistry creates and initializes a new ModuleRegistry instance.

func (*ModuleRegistry) AddModule

func (mr *ModuleRegistry) AddModule(modulePath, filePath string)

AddModule registers a module in the registry. Automatically indexes both the full module path and the short name.

Parameters:

  • modulePath: fully qualified module path (e.g., "myapp.utils.helpers")
  • filePath: absolute file path (e.g., "/project/myapp/utils/helpers.py")

func (*ModuleRegistry) GetModulePath

func (mr *ModuleRegistry) GetModulePath(modulePath string) (string, bool)

GetModulePath returns the file path for a given module, if it exists.

Parameters:

  • modulePath: fully qualified module path

Returns:

  • file path and true if found, empty string and false otherwise

type ORMPattern

type ORMPattern struct {
	Name        string   // Pattern name (e.g., "Django ORM")
	MethodNames []string // Common ORM method names
	Description string   // Human-readable description
}

ORMPattern represents a recognized ORM pattern (e.g., Django ORM, SQLAlchemy). These patterns are dynamically generated at runtime and won't be found in source code, but we can still resolve them by recognizing the pattern.

type Pattern

type Pattern struct {
	ID          string      // Unique identifier (e.g., "SQL-INJECTION-001")
	Name        string      // Human-readable name
	Description string      // What this pattern detects
	Type        PatternType // Pattern category
	Severity    Severity    // Risk level

	// Sources are function names that introduce tainted data
	Sources []string

	// Sinks are function names that consume tainted data dangerously
	Sinks []string

	// Sanitizers are function names that clean tainted data
	Sanitizers []string

	// DangerousFunctions for PatternTypeDangerousFunction
	DangerousFunctions []string

	CWE   string // Common Weakness Enumeration
	OWASP string // OWASP Top 10 category
}

Pattern represents a security pattern to detect in the call graph.

type PatternMatch

type PatternMatch struct {
	PatternID   string   // Pattern identifier
	PatternName string   // Human-readable name
	Description string   // What was detected
	Severity    Severity // Risk level
	CWE         string   // CWE identifier
	OWASP       string   // OWASP category

	// Vulnerability location details
	SourceFQN  string // Fully qualified name of the source function
	SourceCall string // The actual dangerous call (e.g., "input", "request.GET")
	SourceFile string // File path where source is located
	SourceLine uint32 // Line number of source function
	SourceCode string // Code snippet of source function

	SinkFQN  string // Fully qualified name of the sink function
	SinkCall string // The actual dangerous call (e.g., "eval", "exec")
	SinkFile string // File path where sink is located
	SinkLine uint32 // Line number of sink function
	SinkCode string // Code snippet of sink function

	DataFlowPath []string // Complete path from source to sink (FQNs)
}

PatternMatch represents a detected security pattern in the code.

func AnalyzePatterns

func AnalyzePatterns(callGraph *CallGraph, patternRegistry *PatternRegistry) []PatternMatch

AnalyzePatterns runs pattern matching against the call graph. Returns a list of matched patterns with their details.

type PatternMatchDetails

type PatternMatchDetails struct {
	Matched      bool
	SourceFQN    string   // Fully qualified name of function containing the source call
	SourceCall   string   // The actual dangerous call (e.g., "input", "request.GET")
	SinkFQN      string   // Fully qualified name of function containing the sink call
	SinkCall     string   // The actual dangerous call (e.g., "eval", "exec")
	DataFlowPath []string // Complete path from source to sink
}

PatternMatchDetails contains detailed information about a pattern match.

type PatternRegistry

type PatternRegistry struct {
	Patterns       map[string]*Pattern        // Pattern ID -> Pattern
	PatternsByType map[PatternType][]*Pattern // Type -> Patterns
}

PatternRegistry manages security patterns.

func NewPatternRegistry

func NewPatternRegistry() *PatternRegistry

NewPatternRegistry creates a new pattern registry.

func (*PatternRegistry) AddPattern

func (pr *PatternRegistry) AddPattern(pattern *Pattern)

AddPattern registers a pattern in the registry.

func (*PatternRegistry) GetPattern

func (pr *PatternRegistry) GetPattern(id string) (*Pattern, bool)

GetPattern retrieves a pattern by ID.

func (*PatternRegistry) GetPatternsByType

func (pr *PatternRegistry) GetPatternsByType(patternType PatternType) []*Pattern

GetPatternsByType retrieves all patterns of a specific type.

func (*PatternRegistry) LoadDefaultPatterns

func (pr *PatternRegistry) LoadDefaultPatterns()

LoadDefaultPatterns loads the hardcoded example pattern. Additional patterns will be loaded from queries in future PRs.

func (*PatternRegistry) MatchPattern

func (pr *PatternRegistry) MatchPattern(pattern *Pattern, callGraph *CallGraph) *PatternMatchDetails

MatchPattern checks if a call graph matches a pattern. Returns detailed match information if a vulnerability is found.

type PatternType

type PatternType string

PatternType categorizes security patterns for analysis.

const (
	// PatternTypeSourceSink detects tainted data flow from source to sink.
	PatternTypeSourceSink PatternType = "source-sink"

	// PatternTypeMissingSanitizer detects missing sanitization between source and sink.
	PatternTypeMissingSanitizer PatternType = "missing-sanitizer"

	// PatternTypeDangerousFunction detects calls to dangerous functions.
	PatternTypeDangerousFunction PatternType = "dangerous-function"
)

type ReturnStatement

type ReturnStatement struct {
	FunctionFQN string
	ReturnType  *TypeInfo
	Location    Location
}

ReturnStatement represents a return statement in a function.

func ExtractReturnTypes

func ExtractReturnTypes(
	filePath string,
	sourceCode []byte,
	modulePath string,
	builtinRegistry *BuiltinRegistry,
) ([]*ReturnStatement, error)

ExtractReturnTypes analyzes return statements in all functions in a file.

type Severity

type Severity string

Severity indicates the risk level of a security pattern match.

const (
	SeverityCritical Severity = "critical"
	SeverityHigh     Severity = "high"
	SeverityMedium   Severity = "medium"
	SeverityLow      Severity = "low"
)

type TypeInferenceEngine

type TypeInferenceEngine struct {
	Scopes      map[string]*FunctionScope // Function FQN -> scope
	ReturnTypes map[string]*TypeInfo      // Function FQN -> return type
	Builtins    *BuiltinRegistry          // Builtin types registry
	Registry    *ModuleRegistry           // Module registry reference
}

TypeInferenceEngine manages type inference across the codebase. It maintains function scopes, return types, and references to other registries.

func NewTypeInferenceEngine

func NewTypeInferenceEngine(registry *ModuleRegistry) *TypeInferenceEngine

NewTypeInferenceEngine creates a new type inference engine. The engine is initialized with empty scopes and return types.

Parameters:

  • registry: module registry for resolving module paths

Returns:

  • Initialized TypeInferenceEngine

func (*TypeInferenceEngine) AddReturnTypesToEngine

func (te *TypeInferenceEngine) AddReturnTypesToEngine(returnTypes map[string]*TypeInfo)

AddReturnTypesToEngine populates TypeInferenceEngine with return types.

func (*TypeInferenceEngine) AddScope

func (te *TypeInferenceEngine) AddScope(scope *FunctionScope)

AddScope adds or updates a function scope in the engine.

Parameters:

  • scope: the function scope to add

func (*TypeInferenceEngine) GetScope

func (te *TypeInferenceEngine) GetScope(functionFQN string) *FunctionScope

GetScope retrieves a function scope by its fully qualified name.

Parameters:

  • functionFQN: fully qualified name of the function

Returns:

  • FunctionScope if found, nil otherwise

func (*TypeInferenceEngine) ResolveVariableType

func (te *TypeInferenceEngine) ResolveVariableType(
	assignedFrom string,
	confidence float32,
) *TypeInfo

ResolveVariableType resolves the type of a variable assignment from a function call. It looks up the return type of the called function and propagates it with confidence decay.

Parameters:

  • assignedFrom: Function FQN that was called
  • confidence: Base confidence from assignment

Returns:

  • TypeInfo with propagated type, or nil if function has no return type

func (*TypeInferenceEngine) UpdateVariableBindingsWithFunctionReturns

func (te *TypeInferenceEngine) UpdateVariableBindingsWithFunctionReturns()

UpdateVariableBindingsWithFunctionReturns resolves "call:funcName" placeholders. It iterates through all scopes and replaces placeholder types with actual return types.

This enables inter-procedural type propagation:

user = create_user()  # Initially typed as "call:create_user"
# After update, typed as "test.User" based on create_user's return type

type TypeInfo

type TypeInfo struct {
	TypeFQN    string  // Fully qualified type name (e.g., "builtins.str", "myapp.models.User")
	Confidence float32 // Confidence level from 0.0 to 1.0 (1.0 = certain, 0.5 = heuristic, 0.0 = unknown)
	Source     string  // How the type was inferred (e.g., "literal", "assignment", "annotation")
}

TypeInfo represents inferred type information for a variable or expression. It tracks the fully qualified type name, confidence level, and how the type was inferred.

func ResolveClassInstantiation

func ResolveClassInstantiation(
	callNode *sitter.Node,
	sourceCode []byte,
	modulePath string,
	importMap *ImportMap,
	registry *ModuleRegistry,
) *TypeInfo

ResolveClassInstantiation attempts to resolve class instantiation patterns.

type VariableBinding

type VariableBinding struct {
	VarName      string    // Variable name
	Type         *TypeInfo // Inferred type information
	AssignedFrom string    // FQN of function that assigned this value (if from function call)
	Location     Location  // Source location of the assignment
}

VariableBinding tracks a variable's type within a scope. It captures the variable name, its inferred type, and source location.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL