Dataflow analysis helps answer a practical question: how does input data reach an operation that executes SQL, starts a command, or writes a file? CodeGraph searches the CPG for paths from a source to a sink and shows calls, code locations, and detected sanitization functions.
Use the result to review input handling, audit security, or check a fix. Follow the reported path, inspect data transformations, and compare them with the application’s execution conditions.
Choose an analysis route
| Task | Analysis route | What to inspect |
|---|---|---|
| Collect findings for a project | Broad audit with the deep_security profile |
Results from several checks and their execution details. |
| Investigate a selected method | MCP analysis with an exact method_name |
Paths crossing the method and their source and sink functions. |
| Review input handling | Review a reported path | Assignments, calls, sanitization, and trust boundaries. |
Broad project audit
Install the CLI using the CLI guide. Run this PowerShell example from the CodeGraph repository root. You need an accessible GoCPG service and a current CPG for a registered project. CodeGraph uses the project, database, and source association to check freshness. Choose that project’s database and a new report file:
$env:CODEGRAPH_ALLOW_LOCAL_PYTHON_CLI = '1'
python -m src.cli audit --db 'C:/analysis/project.duckdb' `
--profile deep_security --skip-persistence --skip-llm-conclusion --language en `
--format json --output 'dataflow-audit.json'
if ($LASTEXITCODE -ne 0) { throw 'Audit failed' }
$report = Get-Content -Raw 'dataflow-audit.json' | ConvertFrom-Json
The report is written to the current directory and overwrites a file with the same name.
Inspect freshness and analysis-completeness diagnostics. --skip-persistence skips audit projections,
dashboard snapshots, and command history; it does not reduce scope. The output file is still written.
--skip-llm-conclusion produces the report without an LLM conclusion. To include
an LLM summary, configure a provider and remove this option. Inspect each check’s
results and execution details in the report.
For Could not resolve a configured project/source context, choose a registered
project’s database using the import guide. For a stale graph,
refresh the CPG through the project workflow and repeat the audit.
Targeted taint path
For a project using the python_generic domain, the current MCP analysis returns
an empty path list: the analyzer does not load this domain’s source and sink
definitions. For that project, use the SQLite example below and review the source
code. The following section describes the call contract for domains with loaded definitions.
Verify availability and invocation policy for codegraph_security_taint_analysis_run in the active
profile. It requires method_name; defaults are source_category="user_input" and sink_category="sql". The
supported source categories are user_input, file_read, network, and env. Supported sink
categories are sql, command, file_write, xss, and buffer.
Choose sink_category for the operation you are investigating: for example,
sql for SQL queries or command for command execution. This category selects
the search; an empty sink_category uses source_category instead. Sources come
from the domain plugin catalog. The response retains source_category as a call argument.
The result includes paths where the exact method name matches the source, sink, or an intermediate function. Path-count and time limits apply to the overall search before filtering by method. To continue the investigation, check the method name, selected sink category, and limit diagnostics.
MCP calls use the current project context. Select a profile that permits the tool and pass the method and categories shown below. The project context supplies the CPG location.
The tool uses GoCPG gRPC. See the MCP operator guide for setup.
After selecting the correct project, call codegraph_security_taint_analysis_run with these arguments:
{
"method_name": "<exact-method-name-from-your-CPG>",
"source_category": "user_input",
"sink_category": "sql"
}
Replace method_name with the exact name recorded in the current graph. This JSON is the argument object for an MCP call, not a REST request or a shell command. Inspect each returned path and its source/sink functions; use count as the number of returned paths, and record a verdict for each path
after checking execution conditions.
The response contains method, the supplied categories, taint_paths, and count.
The response echoes the supplied categories; analysis selection is described above.
For an Error: ... response, restore access to the selected CPG using the error
message and repeat the call.
Read a path
| Path field | Reviewer action |
|---|---|
source, sink |
Open the functions, files, and lines; check the input and sensitive operation. |
path, functions_crossed |
Inspect the function sequence between source and sink. |
sanitizers |
Check detected sanitization functions against the sink’s input requirements. |
risk_score, risk_level, confidence |
Use analyzer estimates to prioritize review. |
inter_procedural, tainted_fields |
Inspect function boundaries and tracked data fields. |
Validate a finding
For example, when reviewing a SQL path, follow an HTTP request parameter to the argument used to execute a database query. Open the intermediate functions: where is the value assigned, validated, and passed onward? Then inspect the sink call: how is the SQL constructed, and how is the parameter value supplied? If the path includes a sanitization function, check what it does to this data. This is an illustrative review example, rather than analyzer output for your project.
Example: SQL text and a query parameter
Save this code as dataflow-example.py and run python dataflow-example.py
in a terminal with Python 3. The sqlite3 module is part of Python’s standard
library. The example creates an in-memory database and leaves project files
and the deployment database unchanged.
import sqlite3
connection = sqlite3.connect(":memory:")
connection.execute("CREATE TABLE accounts (name TEXT)")
connection.executemany("INSERT INTO accounts VALUES (?)", [("alice",), ("bob",)])
user_input = "alice' OR '1'='1"
unsafe_sql = "SELECT name FROM accounts WHERE name = '" + user_input + "'"
unsafe_rows = connection.execute(unsafe_sql).fetchall()
safe_rows = connection.execute(
"SELECT name FROM accounts WHERE name = ?", (user_input,)
).fetchall()
print("concatenation:", unsafe_rows)
print("parameters:", safe_rows)
connection.close()
Expected output:
concatenation: [('alice',), ('bob',)]
parameters: []
The first query makes the input part of the SQL text. Its OR '1'='1'
condition selects both records. The second query passes the same input as
an argument for the ? parameter. SQLite searches for the literal name
alice' OR '1'='1 and finds no record.
When reviewing a path, identify the source (user_input), the transformation
(the construction of unsafe_sql), and the sink (connection.execute).
Then check whether the application passes input as a separate parameter or
inserts it into the SQL text. The parameterized version preserves the query’s
meaning for this input. This is an educational SQLite example; inspect a path
reported by CodeGraph against your application’s source.
- Confirm the source is actually attacker-controlled in this execution context.
- Follow each call and assignment in the reported path.
- Check sanitization, validation, encoding, authorization, and trust-boundary transitions.
- Verify whether the sink is reachable with the reported data shape.
- Record the code revision, source/sink categories, evidence refs, and reviewer verdict.
After reviewing the reported paths, continue with the application’s remaining sources, sinks, and runtime boundaries.
Reference material
The CLI guide describes how to run a broader audit.