use crate::impl_rule;
use crate::rules::common::{get_source_line, make_finding, walk_tree};
use crate::rules::FileContext;
use crate::{Finding, Language, Severity};
use regex::Regex;
use std::borrow::Cow;
fn resolve_callee<'a>(func_text: &'a str, ctx: &'a FileContext<'_>) -> Cow<'a, str> {
match ctx.python_aliases {
Some(aliases) => aliases.resolve(func_text),
None => Cow::Borrowed(func_text),
}
}
pub struct NoEval;
impl_rule! {
NoEval,
id = "py/no-eval",
severity = Severity::Critical,
cwe = Some("CWE-95"),
description = "Use of eval()/exec() allows arbitrary code execution",
language = Language::Python,
fn check_with_context(_self, source, tree, ctx) {
let mut findings = Vec::new();
walk_tree(tree.root_node(), source, &mut |node, src| {
if node.kind() == "call" {
if let Some(func) = node.child_by_field_name("function") {
let func_text = &src[func.byte_range()];
let resolved = resolve_callee(func_text, ctx);
if resolved.as_ref() == "eval" || resolved.as_ref() == "exec" {
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
&format!(
"{}() allows arbitrary code execution — avoid using it with untrusted input",
resolved
),
node,
src,
));
}
}
}
});
findings
}
}
pub struct NoHardcodedSecret;
impl_rule! {
NoHardcodedSecret,
id = "py/no-hardcoded-secret",
severity = Severity::High,
cwe = Some("CWE-798"),
description = "Hardcoded secret or credential detected",
language = Language::Python,
fn check(_self, source, tree) {
let mut findings = Vec::new();
let secret_pattern =
Regex::new(r"(?i)(password|secret|api_?key|token|auth|credential|private_?key)")
.unwrap();
walk_tree(tree.root_node(), source, &mut |node, src| {
if node.kind() == "assignment" {
if let (Some(left), Some(right)) = (
node.child_by_field_name("left"),
node.child_by_field_name("right"),
) {
let left_text = &src[left.byte_range()];
if secret_pattern.is_match(left_text) && right.kind() == "string" {
let val = &src[right.byte_range()];
let inner = val
.trim_start_matches("f\"")
.trim_start_matches("f'")
.trim_matches(|c| c == '"' || c == '\'');
if inner.len() >= 4 {
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
&format!(
"Hardcoded secret in '{}' — use environment variables or a secrets manager",
left_text
),
node,
src,
));
}
}
}
}
});
findings
}
}
pub struct NoSqlInjection;
impl_rule! {
NoSqlInjection,
id = "py/no-sql-injection",
severity = Severity::Critical,
cwe = Some("CWE-89"),
description = "Potential SQL injection via string formatting",
language = Language::Python,
fn check(_self, source, tree) {
let mut findings = Vec::new();
let sql_pattern =
Regex::new(r"(?i)(SELECT\s+.{0,40}\s+FROM|INSERT\s+INTO|UPDATE\s+.{0,40}\s+SET|DELETE\s+FROM|DROP\s+TABLE|ALTER\s+TABLE|CREATE\s+TABLE|EXEC\s+)").unwrap();
walk_tree(tree.root_node(), source, &mut |node, src| {
if node.kind() == "string" {
let text = &src[node.byte_range()];
if (text.starts_with("f\"")
|| text.starts_with("f'")
|| text.starts_with("f\"\"\""))
&& sql_pattern.is_match(text)
{
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
"SQL query built with f-string — use parameterized queries",
node,
src,
));
}
}
if node.kind() == "binary_operator" {
if let Some(op) = node.child_by_field_name("operator") {
if &src[op.byte_range()] == "%" {
if let Some(left) = node.child_by_field_name("left") {
if left.kind() == "string" {
let text = &src[left.byte_range()];
if sql_pattern.is_match(text) {
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
"SQL query built with % formatting — use parameterized queries",
node,
src,
));
}
}
}
}
}
}
if node.kind() == "call" {
if let Some(func) = node.child_by_field_name("function") {
if func.kind() == "attribute" {
if let Some(attr) = func.child_by_field_name("attribute") {
if &src[attr.byte_range()] == "format" {
if let Some(obj) = func.child_by_field_name("object") {
if obj.kind() == "string" {
let text = &src[obj.byte_range()];
if sql_pattern.is_match(text) {
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
"SQL query built with .format() — use parameterized queries",
node,
src,
));
}
}
}
}
}
}
}
}
if node.kind() == "binary_operator" {
if let Some(op) = node.child_by_field_name("operator") {
if &src[op.byte_range()] == "+" {
if let Some(left) = node.child_by_field_name("left") {
if left.kind() == "string" {
let text = &src[left.byte_range()];
if sql_pattern.is_match(text) {
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
"SQL query built with string concatenation — use parameterized queries",
node,
src,
));
}
}
}
}
}
}
});
findings
}
}
pub struct NoCommandInjection;
impl_rule! {
NoCommandInjection,
id = "py/no-command-injection",
severity = Severity::Critical,
cwe = Some("CWE-78"),
description = "Potential command injection via os.system/subprocess with user input",
language = Language::Python,
fn check_with_context(_self, source, tree, ctx) {
let mut findings = Vec::new();
let dangerous_fns = [
"os.system",
"os.popen",
"subprocess.call",
"subprocess.run",
"subprocess.Popen",
"subprocess.check_output",
"subprocess.check_call",
];
walk_tree(tree.root_node(), source, &mut |node, src| {
if node.kind() == "call" {
if let Some(func) = node.child_by_field_name("function") {
let func_text = &src[func.byte_range()];
let resolved = resolve_callee(func_text, ctx);
if dangerous_fns.contains(&resolved.as_ref()) {
if let Some(args) = node.child_by_field_name("arguments") {
if let Some(first_arg) = args.named_child(0) {
let is_dynamic = match first_arg.kind() {
"string" => {
let text = &src[first_arg.byte_range()];
text.starts_with("f\"") || text.starts_with("f'")
}
"concatenated_string"
| "binary_operator"
| "identifier"
| "call" => true,
_ => false,
};
if is_dynamic {
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
&format!(
"{}() called with dynamic argument — risk of command injection",
resolved
),
node,
src,
));
}
}
}
}
}
}
});
findings
}
}
pub struct NoPathTraversal;
impl_rule! {
NoPathTraversal,
id = "py/no-path-traversal",
severity = Severity::High,
cwe = Some("CWE-22"),
description = "Potential path traversal via open() with user input",
language = Language::Python,
fn check_with_context(_self, source, tree, ctx) {
let mut findings = Vec::new();
walk_tree(tree.root_node(), source, &mut |node, src| {
if node.kind() == "call" {
if let Some(func) = node.child_by_field_name("function") {
let func_text = &src[func.byte_range()];
let resolved = resolve_callee(func_text, ctx);
let sink_fns = ["open", "os.remove", "os.unlink", "os.listdir", "os.scandir"];
if sink_fns.contains(&resolved.as_ref()) {
if let Some(args) = node.child_by_field_name("arguments") {
if let Some(first_arg) = args.named_child(0) {
let is_dynamic = match first_arg.kind() {
"binary_operator" | "concatenated_string" | "identifier" => {
true
}
"string" => {
let text = &src[first_arg.byte_range()];
text.starts_with("f\"") || text.starts_with("f'")
}
_ => false,
};
if is_dynamic {
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
&format!(
"{}() called with dynamic path — validate and sanitize to prevent path traversal",
resolved
),
node,
src,
));
}
}
}
}
}
}
});
findings
}
}
pub struct NoSsrf;
impl_rule! {
NoSsrf,
id = "py/no-ssrf",
severity = Severity::High,
cwe = Some("CWE-918"),
description = "Potential SSRF via dynamic outbound request URL",
language = Language::Python,
fn check_with_context(_self, source, tree, ctx) {
let mut findings = Vec::new();
let request_fns = [
"requests.get",
"requests.post",
"requests.put",
"requests.delete",
"requests.head",
"requests.patch",
"requests.request",
"httpx.get",
"httpx.post",
"httpx.put",
"httpx.delete",
"httpx.request",
"urllib.request.urlopen",
];
walk_tree(tree.root_node(), source, &mut |node, src| {
if node.kind() != "call" {
return;
}
let Some(func) = node.child_by_field_name("function") else {
return;
};
let func_text = &src[func.byte_range()];
let resolved = resolve_callee(func_text, ctx);
if !request_fns.contains(&resolved.as_ref()) {
return;
}
let Some(args) = node.child_by_field_name("arguments") else {
return;
};
let url_arg = if resolved.as_ref() == "requests.request"
|| resolved.as_ref() == "httpx.request"
{
args.named_child(1)
} else {
args.named_child(0)
};
let Some(url_arg) = url_arg else {
return;
};
let is_dynamic = match url_arg.kind() {
"string" => {
let text = &src[url_arg.byte_range()];
text.starts_with("f\"") || text.starts_with("f'")
}
"identifier" | "call" | "subscript" | "attribute" | "binary_operator" => true,
_ => false,
};
if is_dynamic {
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
&format!(
"{} called with dynamic URL — validate and allowlist outbound destinations to prevent SSRF",
resolved
),
node,
src,
));
}
});
findings
}
}
pub struct NoWeakCrypto;
impl_rule! {
NoWeakCrypto,
id = "py/no-weak-crypto",
severity = Severity::Medium,
cwe = Some("CWE-327"),
description = "Use of weak cryptographic hash (MD5/SHA1)",
language = Language::Python,
fn check_with_context(_self, source, tree, ctx) {
let mut findings = Vec::new();
walk_tree(tree.root_node(), source, &mut |node, src| {
if node.kind() == "call" {
if let Some(func) = node.child_by_field_name("function") {
let func_text = &src[func.byte_range()];
let resolved = resolve_callee(func_text, ctx);
if resolved.as_ref() == "hashlib.md5" || resolved.as_ref() == "hashlib.sha1" {
let algo = if resolved.as_ref().contains("md5") {
"MD5"
} else {
"SHA1"
};
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
&format!(
"hashlib.{}() is cryptographically weak — use sha256 or stronger",
algo.to_lowercase()
),
node,
src,
));
}
if resolved.as_ref() == "hashlib.new" {
if let Some(args) = node.child_by_field_name("arguments") {
if let Some(first_arg) = args.named_child(0) {
if first_arg.kind() == "string" {
let val = &src[first_arg.byte_range()];
let inner = val.trim_matches(|c| c == '"' || c == '\'');
if inner == "md5" || inner == "sha1" {
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
&format!(
"hashlib.new('{}') is cryptographically weak — use sha256 or stronger",
inner
),
node,
src,
));
}
}
}
}
}
}
}
});
findings
}
}
pub struct NoPickle;
impl_rule! {
NoPickle,
id = "py/no-pickle",
severity = Severity::High,
cwe = Some("CWE-502"),
description = "Deserialization of untrusted data via pickle",
language = Language::Python,
fn check_with_context(_self, source, tree, ctx) {
let mut findings = Vec::new();
let dangerous_fns = [
"pickle.loads",
"pickle.load",
"cPickle.loads",
"cPickle.load",
];
walk_tree(tree.root_node(), source, &mut |node, src| {
if node.kind() == "call" {
if let Some(func) = node.child_by_field_name("function") {
let func_text = &src[func.byte_range()];
let resolved = resolve_callee(func_text, ctx);
if dangerous_fns.contains(&resolved.as_ref()) {
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
&format!(
"{}() deserializes untrusted data — can execute arbitrary code",
resolved
),
node,
src,
));
}
}
}
});
findings
}
}
pub struct NoYamlLoad;
impl_rule! {
NoYamlLoad,
id = "py/no-yaml-load",
severity = Severity::High,
cwe = Some("CWE-502"),
description = "yaml.load() without SafeLoader can execute arbitrary code",
language = Language::Python,
fn check_with_context(_self, source, tree, ctx) {
let mut findings = Vec::new();
walk_tree(tree.root_node(), source, &mut |node, src| {
if node.kind() == "call" {
if let Some(func) = node.child_by_field_name("function") {
let func_text = &src[func.byte_range()];
let resolved = resolve_callee(func_text, ctx);
if resolved.as_ref() == "yaml.load" {
if let Some(args) = node.child_by_field_name("arguments") {
let args_text = &src[args.byte_range()];
if !args_text.contains("SafeLoader")
&& !args_text.contains("safe_load")
&& !args_text.contains("BaseLoader")
{
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
"yaml.load() without SafeLoader — use yaml.safe_load() or pass Loader=SafeLoader",
node,
src,
));
}
}
}
}
}
});
findings
}
}
pub struct NoDebugTrue;
impl_rule! {
NoDebugTrue,
id = "py/no-debug-true",
severity = Severity::Medium,
cwe = Some("CWE-489"),
description = "DEBUG = True left enabled — disable in production",
language = Language::Python,
fn check(_self, source, tree) {
let mut findings = Vec::new();
walk_tree(tree.root_node(), source, &mut |node, src| {
if node.kind() == "assignment" {
if let (Some(left), Some(right)) = (
node.child_by_field_name("left"),
node.child_by_field_name("right"),
) {
let left_text = &src[left.byte_range()];
let right_text = &src[right.byte_range()];
if left_text == "DEBUG" && right_text == "True" {
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
"DEBUG = True — ensure debug mode is disabled in production (Django CWE-489)",
node,
src,
));
}
}
}
});
findings
}
}
pub struct FlaskDebugMode;
impl_rule! {
FlaskDebugMode,
id = "py/flask-debug-mode",
severity = Severity::High,
cwe = Some("CWE-489"),
description = "Flask app.run(debug=True) exposes debugger and reloader in production",
language = Language::Python,
fn check(_self, source, tree) {
let mut findings = Vec::new();
walk_tree(tree.root_node(), source, &mut |node, src| {
if node.kind() == "call" {
if let Some(func) = node.child_by_field_name("function") {
if func.kind() == "attribute" {
if let Some(attr) = func.child_by_field_name("attribute") {
if &src[attr.byte_range()] == "run" {
if let Some(args) = node.child_by_field_name("arguments") {
let args_text = &src[args.byte_range()];
if args_text.contains("debug=True")
|| args_text.contains("debug = True")
{
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
"Flask app.run(debug=True) — exposes Werkzeug debugger, disable in production",
node,
src,
));
}
}
}
}
}
}
}
if node.kind() == "assignment" {
if let (Some(left), Some(right)) = (
node.child_by_field_name("left"),
node.child_by_field_name("right"),
) {
let left_text = &src[left.byte_range()];
let right_text = &src[right.byte_range()];
if left_text.ends_with(".debug") && right_text == "True" {
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
"app.debug = True — exposes debugger, disable in production",
node,
src,
));
}
}
}
});
findings
}
}
pub struct DjangoSecretKeyHardcoded;
impl_rule! {
DjangoSecretKeyHardcoded,
id = "py/django-secret-key-hardcoded",
severity = Severity::High,
cwe = Some("CWE-798"),
description = "Django SECRET_KEY hardcoded in source code",
language = Language::Python,
fn check(_self, source, tree) {
let mut findings = Vec::new();
walk_tree(tree.root_node(), source, &mut |node, src| {
if node.kind() == "assignment" {
if let (Some(left), Some(right)) = (
node.child_by_field_name("left"),
node.child_by_field_name("right"),
) {
let left_text = &src[left.byte_range()];
if left_text == "SECRET_KEY" && right.kind() == "string" {
let val = &src[right.byte_range()];
let inner = val.trim_matches(|c| c == '"' || c == '\'');
if inner.len() >= 4 {
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
"Django SECRET_KEY is hardcoded — use an environment variable or secrets manager",
node,
src,
));
}
}
}
}
});
findings
}
}
pub struct NoOpenRedirect;
impl_rule! {
NoOpenRedirect,
id = "py/no-open-redirect",
severity = Severity::Medium,
cwe = Some("CWE-601"),
description = "Open redirect via redirect() with user-controlled input",
language = Language::Python,
fn check(_self, source, tree) {
let mut findings = Vec::new();
let redirect_fns = ["redirect", "HttpResponseRedirect"];
walk_tree(tree.root_node(), source, &mut |node, src| {
if node.kind() == "call" {
if let Some(func) = node.child_by_field_name("function") {
let func_text = &src[func.byte_range()];
let func_name = func_text.rsplit('.').next().unwrap_or(func_text);
if redirect_fns.contains(&func_name) {
if let Some(args) = node.child_by_field_name("arguments") {
if let Some(first_arg) = args.named_child(0) {
let is_dynamic = match first_arg.kind() {
"string" => {
let text = &src[first_arg.byte_range()];
text.starts_with("f\"") || text.starts_with("f'")
}
"identifier" | "call" | "subscript" | "attribute"
| "binary_operator" => true,
_ => false,
};
if is_dynamic {
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
&format!(
"{}() with dynamic URL — validate target to prevent open redirect",
func_name
),
node,
src,
));
}
}
}
}
}
}
});
findings
}
}
pub struct NoCorsStar;
impl_rule! {
NoCorsStar,
id = "py/no-cors-star",
severity = Severity::Medium,
cwe = Some("CWE-942"),
description = "CORS misconfiguration allowing all origins",
language = Language::Python,
fn check(_self, source, tree) {
let mut findings = Vec::new();
walk_tree(tree.root_node(), source, &mut |node, src| {
if node.kind() == "assignment" {
if let (Some(left), Some(right)) = (
node.child_by_field_name("left"),
node.child_by_field_name("right"),
) {
let left_text = &src[left.byte_range()];
let right_text = &src[right.byte_range()];
if (left_text == "CORS_ALLOW_ALL_ORIGINS"
|| left_text == "CORS_ORIGIN_ALLOW_ALL")
&& right_text == "True"
{
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
&format!("{} = True — restrict CORS to specific origins", left_text),
node,
src,
));
}
}
}
if node.kind() == "call" {
if let Some(func) = node.child_by_field_name("function") {
let func_text = &src[func.byte_range()];
if func_text.contains("header") || func_text.contains("Header") {
let node_text = &src[node.byte_range()];
if node_text.contains("Access-Control-Allow-Origin")
&& node_text.contains("*")
{
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
"Access-Control-Allow-Origin set to '*' — restrict to specific origins",
node,
src,
));
}
}
}
}
if node.kind() == "keyword_argument" {
if let (Some(name), Some(value)) = (
node.child_by_field_name("name"),
node.child_by_field_name("value"),
) {
let name_text = &src[name.byte_range()];
if name_text == "allow_origins" || name_text == "origins" {
let value_text = &src[value.byte_range()];
if value_text.contains("\"*\"") || value_text.contains("'*'") {
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
"CORS allow_origins includes '*' — restrict to specific origins",
node,
src,
));
}
}
}
}
});
findings
}
}
pub struct FlaskSecretKeyHardcoded;
impl_rule! {
FlaskSecretKeyHardcoded,
id = "py/flask-secret-key-hardcoded",
severity = Severity::High,
cwe = Some("CWE-798"),
description = "Flask SECRET_KEY hardcoded in source code",
language = Language::Python,
fn check(_self, source, tree) {
let mut findings = Vec::new();
walk_tree(tree.root_node(), source, &mut |node, src| {
if node.kind() != "assignment" {
return;
}
let (Some(left), Some(right)) = (
node.child_by_field_name("left"),
node.child_by_field_name("right"),
) else {
return;
};
if right.kind() != "string" {
return;
}
let left_text = &src[left.byte_range()];
let is_flask_secret = left_text == "app.secret_key"
|| (left_text.contains("config") && left_text.contains("SECRET_KEY"));
if !is_flask_secret {
return;
}
let val = &src[right.byte_range()];
let inner = val.trim_matches(|c| c == '"' || c == '\'');
if inner.len() < 4 {
return;
}
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
"Flask SECRET_KEY is hardcoded — use an environment variable or secrets manager",
node,
src,
));
});
findings
}
}
pub struct SessionCookieSecureDisabled;
impl_rule! {
SessionCookieSecureDisabled,
id = "py/session-cookie-secure-disabled",
severity = Severity::Medium,
cwe = Some("CWE-614"),
description = "SESSION_COOKIE_SECURE disabled in source code",
language = Language::Python,
fn check(_self, source, tree) {
let mut findings = Vec::new();
walk_tree(tree.root_node(), source, &mut |node, src| {
if node.kind() != "assignment" {
return;
}
let (Some(left), Some(right)) = (
node.child_by_field_name("left"),
node.child_by_field_name("right"),
) else {
return;
};
let left_text = &src[left.byte_range()];
let right_text = &src[right.byte_range()];
let is_session_cookie_secure =
left_text == "SESSION_COOKIE_SECURE" || left_text.contains("SESSION_COOKIE_SECURE");
if is_session_cookie_secure && right_text == "False" {
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
"SESSION_COOKIE_SECURE = False — session cookies may be sent over HTTP",
node,
src,
));
}
});
findings
}
}
pub struct SessionCookieHttpOnlyDisabled;
impl_rule! {
SessionCookieHttpOnlyDisabled,
id = "py/session-cookie-httponly-disabled",
severity = Severity::Medium,
cwe = Some("CWE-1004"),
description = "SESSION_COOKIE_HTTPONLY disabled in source code",
language = Language::Python,
fn check(_self, source, tree) {
let mut findings = Vec::new();
walk_tree(tree.root_node(), source, &mut |node, src| {
if node.kind() != "assignment" {
return;
}
let (Some(left), Some(right)) = (
node.child_by_field_name("left"),
node.child_by_field_name("right"),
) else {
return;
};
let left_text = &src[left.byte_range()];
let right_text = &src[right.byte_range()];
let is_session_cookie_httponly = left_text == "SESSION_COOKIE_HTTPONLY"
|| left_text.contains("SESSION_COOKIE_HTTPONLY");
if is_session_cookie_httponly && right_text == "False" {
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
"SESSION_COOKIE_HTTPONLY = False — session cookies may be exposed to client-side scripts",
node,
src,
));
}
});
findings
}
}
pub struct SessionCookieSameSiteDisabled;
impl_rule! {
SessionCookieSameSiteDisabled,
id = "py/session-cookie-samesite-disabled",
severity = Severity::Medium,
cwe = Some("CWE-352"),
description = "SESSION_COOKIE_SAMESITE disabled in source code",
language = Language::Python,
fn check(_self, source, tree) {
let mut findings = Vec::new();
walk_tree(tree.root_node(), source, &mut |node, src| {
if node.kind() != "assignment" {
return;
}
let (Some(left), Some(right)) = (
node.child_by_field_name("left"),
node.child_by_field_name("right"),
) else {
return;
};
let left_text = &src[left.byte_range()];
let right_text = &src[right.byte_range()];
let is_session_cookie_samesite = left_text == "SESSION_COOKIE_SAMESITE"
|| left_text.contains("SESSION_COOKIE_SAMESITE");
let disabled = right_text == "None"
|| right_text == "\"None\""
|| right_text == "'None'"
|| right_text == "\"none\""
|| right_text == "'none'"
|| right_text == "False";
if is_session_cookie_samesite && disabled {
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
"SESSION_COOKIE_SAMESITE disabled — set it to 'Lax' or 'Strict' to reduce CSRF risk",
node,
src,
));
}
});
findings
}
}
pub struct CsrfCookieSecureDisabled;
impl_rule! {
CsrfCookieSecureDisabled,
id = "py/csrf-cookie-secure-disabled",
severity = Severity::Medium,
cwe = Some("CWE-614"),
description = "CSRF_COOKIE_SECURE disabled in source code",
language = Language::Python,
fn check(_self, source, tree) {
let mut findings = Vec::new();
walk_tree(tree.root_node(), source, &mut |node, src| {
if node.kind() != "assignment" {
return;
}
let (Some(left), Some(right)) = (
node.child_by_field_name("left"),
node.child_by_field_name("right"),
) else {
return;
};
let left_text = &src[left.byte_range()];
let right_text = &src[right.byte_range()];
let is_csrf_cookie_secure =
left_text == "CSRF_COOKIE_SECURE" || left_text.contains("CSRF_COOKIE_SECURE");
if is_csrf_cookie_secure && right_text == "False" {
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
"CSRF_COOKIE_SECURE = False — CSRF cookies may be sent over HTTP",
node,
src,
));
}
});
findings
}
}
pub struct CsrfCookieHttpOnlyDisabled;
impl_rule! {
CsrfCookieHttpOnlyDisabled,
id = "py/csrf-cookie-httponly-disabled",
severity = Severity::Medium,
cwe = Some("CWE-1004"),
description = "CSRF_COOKIE_HTTPONLY disabled in source code",
language = Language::Python,
fn check(_self, source, tree) {
let mut findings = Vec::new();
walk_tree(tree.root_node(), source, &mut |node, src| {
if node.kind() != "assignment" {
return;
}
let (Some(left), Some(right)) = (
node.child_by_field_name("left"),
node.child_by_field_name("right"),
) else {
return;
};
let left_text = &src[left.byte_range()];
let right_text = &src[right.byte_range()];
let is_csrf_cookie_httponly =
left_text == "CSRF_COOKIE_HTTPONLY" || left_text.contains("CSRF_COOKIE_HTTPONLY");
if is_csrf_cookie_httponly && right_text == "False" {
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
"CSRF_COOKIE_HTTPONLY = False — CSRF cookies may be exposed to client-side scripts",
node,
src,
));
}
});
findings
}
}
pub struct CsrfCookieSameSiteDisabled;
impl_rule! {
CsrfCookieSameSiteDisabled,
id = "py/csrf-cookie-samesite-disabled",
severity = Severity::Medium,
cwe = Some("CWE-352"),
description = "CSRF_COOKIE_SAMESITE disabled in source code",
language = Language::Python,
fn check(_self, source, tree) {
let mut findings = Vec::new();
walk_tree(tree.root_node(), source, &mut |node, src| {
if node.kind() != "assignment" {
return;
}
let (Some(left), Some(right)) = (
node.child_by_field_name("left"),
node.child_by_field_name("right"),
) else {
return;
};
let left_text = &src[left.byte_range()];
let right_text = &src[right.byte_range()];
let is_csrf_cookie_samesite =
left_text == "CSRF_COOKIE_SAMESITE" || left_text.contains("CSRF_COOKIE_SAMESITE");
let disabled = right_text == "None"
|| right_text == "\"None\""
|| right_text == "'None'"
|| right_text == "\"none\""
|| right_text == "'none'"
|| right_text == "False";
if is_csrf_cookie_samesite && disabled {
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
"CSRF_COOKIE_SAMESITE disabled — set it to 'Lax' or 'Strict' to reduce CSRF risk",
node,
src,
));
}
});
findings
}
}
pub struct CsrfExempt;
impl_rule! {
CsrfExempt,
id = "py/csrf-exempt",
severity = Severity::High,
cwe = Some("CWE-352"),
description = "View marked csrf_exempt",
language = Language::Python,
fn check(_self, source, tree) {
let mut findings = Vec::new();
walk_tree(tree.root_node(), source, &mut |node, src| {
let text = &src[node.byte_range()];
if node.kind() == "decorator" && text.contains("csrf_exempt") {
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
"@csrf_exempt disables CSRF protection — prefer scoped exemptions or validated alternative controls",
node,
src,
));
}
});
findings
}
}
pub struct WtfCsrfDisabled;
impl_rule! {
WtfCsrfDisabled,
id = "py/wtf-csrf-disabled",
severity = Severity::High,
cwe = Some("CWE-352"),
description = "Flask-WTF CSRF protection disabled in source code",
language = Language::Python,
fn check(_self, source, tree) {
let mut findings = Vec::new();
walk_tree(tree.root_node(), source, &mut |node, src| {
if node.kind() != "assignment" {
return;
}
let (Some(left), Some(right)) = (
node.child_by_field_name("left"),
node.child_by_field_name("right"),
) else {
return;
};
let left_text = &src[left.byte_range()];
let right_text = &src[right.byte_range()];
if left_text.contains("WTF_CSRF_ENABLED") && right_text == "False" {
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
"Flask-WTF CSRF protection disabled — keep WTF_CSRF_ENABLED enabled",
node,
src,
));
}
});
findings
}
}
pub struct WtfCsrfCheckDefaultDisabled;
impl_rule! {
WtfCsrfCheckDefaultDisabled,
id = "py/wtf-csrf-check-default-disabled",
severity = Severity::High,
cwe = Some("CWE-352"),
description = "Flask-WTF default CSRF checks disabled in source code",
language = Language::Python,
fn check(_self, source, tree) {
let mut findings = Vec::new();
walk_tree(tree.root_node(), source, &mut |node, src| {
if node.kind() != "assignment" {
return;
}
let (Some(left), Some(right)) = (
node.child_by_field_name("left"),
node.child_by_field_name("right"),
) else {
return;
};
let left_text = &src[left.byte_range()];
let right_text = &src[right.byte_range()];
if left_text.contains("WTF_CSRF_CHECK_DEFAULT") && right_text == "False" {
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
"Flask-WTF default CSRF checks disabled — keep WTF_CSRF_CHECK_DEFAULT enabled",
node,
src,
));
}
});
findings
}
}
pub struct DjangoAllowedHostsWildcard;
impl_rule! {
DjangoAllowedHostsWildcard,
id = "py/django-allowed-hosts-wildcard",
severity = Severity::Medium,
cwe = Some("CWE-346"),
description = "Django ALLOWED_HOSTS allows all hosts",
language = Language::Python,
fn check(_self, source, tree) {
let mut findings = Vec::new();
walk_tree(tree.root_node(), source, &mut |node, src| {
if node.kind() != "assignment" {
return;
}
let (Some(left), Some(right)) = (
node.child_by_field_name("left"),
node.child_by_field_name("right"),
) else {
return;
};
let left_text = &src[left.byte_range()];
let right_text = &src[right.byte_range()];
if left_text.contains("ALLOWED_HOSTS") && right_text.contains("\"*\"") {
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
"Django ALLOWED_HOSTS contains '*' — restrict hostnames explicitly to reduce host header abuse risk",
node,
src,
));
}
});
findings
}
}
pub struct SecureSslRedirectDisabled;
impl_rule! {
SecureSslRedirectDisabled,
id = "py/secure-ssl-redirect-disabled",
severity = Severity::Medium,
cwe = Some("CWE-319"),
description = "Django SECURE_SSL_REDIRECT disabled in source code",
language = Language::Python,
fn check(_self, source, tree) {
let mut findings = Vec::new();
walk_tree(tree.root_node(), source, &mut |node, src| {
if node.kind() != "assignment" {
return;
}
let (Some(left), Some(right)) = (
node.child_by_field_name("left"),
node.child_by_field_name("right"),
) else {
return;
};
let left_text = &src[left.byte_range()];
let right_text = &src[right.byte_range()];
if left_text.contains("SECURE_SSL_REDIRECT") && right_text == "False" {
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
"SECURE_SSL_REDIRECT = False — enable HTTPS redirect in production-facing Django deployments",
node,
src,
));
}
});
findings
}
}
pub struct JwtNoVerify;
impl_rule! {
JwtNoVerify,
id = "py/jwt-no-verify",
severity = Severity::Critical,
cwe = Some("CWE-347"),
description = "JWT decoded without signature verification",
language = Language::Python,
fn check_with_context(_self, source, tree, ctx) {
let mut findings = Vec::new();
walk_tree(tree.root_node(), source, &mut |node, src| {
if node.kind() != "call" {
return;
}
let Some(func) = node.child_by_field_name("function") else {
return;
};
let func_text = &src[func.byte_range()];
let resolved = resolve_callee(func_text, ctx);
if resolved.as_ref() != "jwt.decode" {
return;
}
let node_text = &src[node.byte_range()];
let no_verify = node_text.contains("verify=False")
|| node_text.contains("verify_signature=False")
|| node_text.contains("\"verify_signature\": False")
|| node_text.contains("'verify_signature': False")
|| node_text.contains("\"verify_signature\":False")
|| node_text.contains("'verify_signature':False");
let none_algo = node_text.contains("algorithms=[\"none\"]")
|| node_text.contains("algorithms=['none']")
|| node_text.contains("algorithms=[\"None\"]")
|| node_text.contains("algorithms=['None']");
if no_verify || none_algo {
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
"JWT decoded without signature verification — always verify tokens with a trusted key",
node,
src,
));
}
});
findings
}
}
pub struct JwtHardcodedSecret;
impl_rule! {
JwtHardcodedSecret,
id = "py/jwt-hardcoded-secret",
severity = Severity::High,
cwe = Some("CWE-798"),
description = "JWT signing or verification with a hardcoded secret",
language = Language::Python,
fn check_with_context(_self, source, tree, ctx) {
let mut findings = Vec::new();
walk_tree(tree.root_node(), source, &mut |node, src| {
if node.kind() != "call" {
return;
}
let Some(func) = node.child_by_field_name("function") else {
return;
};
let func_text = &src[func.byte_range()];
let resolved = resolve_callee(func_text, ctx);
if resolved.as_ref() != "jwt.encode" && resolved.as_ref() != "jwt.decode" {
return;
}
let Some(args) = node.child_by_field_name("arguments") else {
return;
};
let Some(secret_arg) = args.named_child(1) else {
return;
};
if secret_arg.kind() == "string" || secret_arg.kind() == "concatenated_string" {
let secret_text = &src[secret_arg.byte_range()];
let inner = secret_text.trim_matches(|c| c == '"' || c == '\'');
if inner.len() >= 4 {
findings.push(make_finding(
_self.id(),
_self.severity(),
_self.cwe(),
"JWT secret is hardcoded — load signing keys from environment or a secrets manager",
node,
src,
));
}
}
});
findings
}
}
use crate::rules::python_taint::{self, python_taint_sources, NodeMatcher, TaintSpec};
fn call_sink(canonical: &str) -> NodeMatcher {
NodeMatcher::Call {
canonical: canonical.into(),
description: canonical.into(),
}
}
struct TaintRuleMeta<'a> {
rule_id: &'a str,
severity: Severity,
cwe: Option<&'a str>,
fix_suggestion: Option<&'a str>,
}
fn map_taint_findings(
meta: &TaintRuleMeta<'_>,
source: &str,
tree: &tree_sitter::Tree,
ctx: &FileContext<'_>,
spec: &TaintSpec,
format_description: impl Fn(&str, &str) -> String,
) -> Vec<Finding> {
let cross_file_info = match (ctx.cross_file_summaries, ctx.python_import_paths) {
(Some(summaries), Some(import_paths)) => Some(python_taint::CrossFileInfo {
import_to_path: import_paths,
summaries,
current_rule_id: meta.rule_id,
}),
_ => None,
};
let raw = python_taint::analyze_tree_with_cross_file(
tree.root_node(),
source,
spec,
ctx.python_aliases,
cross_file_info.as_ref(),
);
raw.into_iter()
.map(|t| Finding {
rule_id: meta.rule_id.to_string(),
severity: meta.severity,
cwe: meta.cwe.map(|s| s.to_string()),
description: format_description(&t.source_description, &t.sink_description),
file: String::new(),
line: t.sink_line,
column: t.sink_column,
end_line: t.sink_end_line,
end_column: t.sink_end_column,
snippet: get_source_line(source, t.sink_start_byte),
source_line: Some(t.source_line),
source_description: Some(t.source_description),
sink_line: Some(t.sink_line),
sink_description: Some(t.sink_description),
fix_suggestion: meta.fix_suggestion.map(|s| s.to_string()),
})
.collect()
}
pub struct TaintPickleDeserialization;
impl TaintPickleDeserialization {
fn spec() -> TaintSpec {
TaintSpec {
sources: python_taint_sources(),
sinks: vec![
call_sink("pickle.loads"),
call_sink("pickle.load"),
call_sink("cPickle.loads"),
call_sink("cPickle.load"),
],
sanitizers: vec![],
}
}
}
impl_rule! {
TaintPickleDeserialization,
id = "py/taint-pickle-deserialization",
severity = Severity::Critical,
cwe = Some("CWE-502"),
description = "Untrusted input reaches pickle deserialization sink",
language = Language::Python,
fn check_with_context(_self, source, tree, ctx) {
let meta = TaintRuleMeta {
rule_id: _self.id(),
severity: _self.severity(),
cwe: _self.cwe(),
fix_suggestion: Some(
"Use `json` or `msgpack` instead of pickle for untrusted data: `json.loads(data)`",
),
};
map_taint_findings(&meta, source, tree, ctx, &Self::spec(), |src, sink| {
format!(
"{} reaches {} — untrusted input can execute arbitrary code via pickle",
src, sink
)
})
}
}
pub struct TaintEvalFromRequest;
impl TaintEvalFromRequest {
fn spec() -> TaintSpec {
TaintSpec {
sources: python_taint_sources(),
sinks: vec![call_sink("eval"), call_sink("exec")],
sanitizers: vec![],
}
}
}
impl_rule! {
TaintEvalFromRequest,
id = "py/taint-eval",
severity = Severity::Critical,
cwe = Some("CWE-95"),
description = "Untrusted input reaches eval/exec sink",
language = Language::Python,
fn check_with_context(_self, source, tree, ctx) {
let meta = TaintRuleMeta {
rule_id: _self.id(),
severity: _self.severity(),
cwe: _self.cwe(),
fix_suggestion: Some(
"Use `ast.literal_eval()` for safe evaluation, or remove eval/exec entirely",
),
};
map_taint_findings(&meta, source, tree, ctx, &Self::spec(), |src, sink| {
format!(
"{} reaches {} — untrusted input can execute arbitrary Python code",
src, sink
)
})
}
}
pub struct TaintCommandInjectionFromRequest;
impl TaintCommandInjectionFromRequest {
fn spec() -> TaintSpec {
TaintSpec {
sources: python_taint_sources(),
sinks: vec![
call_sink("os.system"),
call_sink("os.popen"),
call_sink("subprocess.run"),
call_sink("subprocess.Popen"),
call_sink("subprocess.call"),
call_sink("subprocess.check_call"),
call_sink("subprocess.check_output"),
],
sanitizers: vec![
call_sink("shlex.quote"),
call_sink("shlex.join"),
call_sink("subprocess.list2cmdline"),
],
}
}
}
impl_rule! {
TaintCommandInjectionFromRequest,
id = "py/taint-command-injection",
severity = Severity::Critical,
cwe = Some("CWE-78"),
description = "Untrusted input reaches OS command execution sink",
language = Language::Python,
fn check_with_context(_self, source, tree, ctx) {
let meta = TaintRuleMeta {
rule_id: _self.id(),
severity: _self.severity(),
cwe: _self.cwe(),
fix_suggestion: Some("Use `shlex.quote()` to escape arguments, or pass a list to `subprocess.run([...])` instead of a shell string"),
};
map_taint_findings(&meta, source, tree, ctx, &Self::spec(), |src, sink| {
format!(
"{} reaches {} — untrusted input can inject OS commands",
src, sink
)
})
}
}
pub struct TaintSsrfFromRequest;
impl TaintSsrfFromRequest {
fn spec() -> TaintSpec {
TaintSpec {
sources: python_taint_sources(),
sinks: vec![
call_sink("urllib.request.urlopen"),
call_sink("requests.get"),
call_sink("requests.post"),
call_sink("requests.put"),
call_sink("requests.delete"),
call_sink("requests.request"),
call_sink("httpx.get"),
call_sink("httpx.post"),
],
sanitizers: vec![],
}
}
}
impl_rule! {
TaintSsrfFromRequest,
id = "py/taint-ssrf",
severity = Severity::High,
cwe = Some("CWE-918"),
description = "Untrusted input reaches outbound HTTP sink (potential SSRF)",
language = Language::Python,
fn check_with_context(_self, source, tree, ctx) {
let meta = TaintRuleMeta {
rule_id: _self.id(),
severity: _self.severity(),
cwe: _self.cwe(),
fix_suggestion: Some(
"Validate URLs against an allowlist of permitted hosts before making requests",
),
};
map_taint_findings(&meta, source, tree, ctx, &Self::spec(), |src, sink| {
format!(
"{} reaches {} — untrusted input can drive server-side request forgery",
src, sink
)
})
}
}
pub struct TaintYamlLoadFromRequest;
impl TaintYamlLoadFromRequest {
fn spec() -> TaintSpec {
TaintSpec {
sources: python_taint_sources(),
sinks: vec![
call_sink("yaml.load"),
call_sink("yaml.unsafe_load"),
call_sink("yaml.full_load"),
],
sanitizers: vec![],
}
}
}
impl_rule! {
TaintYamlLoadFromRequest,
id = "py/taint-yaml-load",
severity = Severity::Critical,
cwe = Some("CWE-502"),
description = "Untrusted input reaches unsafe YAML loader",
language = Language::Python,
fn check_with_context(_self, source, tree, ctx) {
let meta = TaintRuleMeta {
rule_id: _self.id(),
severity: _self.severity(),
cwe: _self.cwe(),
fix_suggestion: Some(
"Use `yaml.safe_load()` instead of `yaml.load()` for untrusted input",
),
};
map_taint_findings(&meta, source, tree, ctx, &Self::spec(), |src, sink| {
format!(
"{} reaches {} — untrusted input can execute arbitrary code via YAML deserialization",
src, sink
)
})
}
}
pub struct TaintSqlInjectionFromRequest;
impl TaintSqlInjectionFromRequest {
fn spec() -> TaintSpec {
TaintSpec {
sources: python_taint_sources(),
sinks: vec![
NodeMatcher::MethodName {
method: "execute".into(),
description: "cursor/connection.execute".into(),
},
NodeMatcher::MethodName {
method: "executemany".into(),
description: "cursor/connection.executemany".into(),
},
NodeMatcher::MethodName {
method: "executescript".into(),
description: "sqlite3.Cursor.executescript".into(),
},
],
sanitizers: vec![call_sink("escape_string"), call_sink("quote_ident")],
}
}
}
impl_rule! {
TaintSqlInjectionFromRequest,
id = "py/taint-sql-injection",
severity = Severity::Critical,
cwe = Some("CWE-89"),
description = "Untrusted input reaches DB execute sink",
language = Language::Python,
fn check_with_context(_self, source, tree, ctx) {
let meta = TaintRuleMeta {
rule_id: _self.id(),
severity: _self.severity(),
cwe: _self.cwe(),
fix_suggestion: Some("Use parameterized queries: `cur.execute(\"SELECT * FROM users WHERE name = ?\", (name,))`"),
};
map_taint_findings(&meta, source, tree, ctx, &Self::spec(), |src, sink| {
format!("{} reaches {} — untrusted input can inject SQL", src, sink)
})
}
}
pub struct TaintSsti;
impl TaintSsti {
fn spec() -> TaintSpec {
TaintSpec {
sources: python_taint_sources(),
sinks: vec![
call_sink("jinja2.Template"),
call_sink("Template"),
call_sink("flask.render_template_string"),
call_sink("render_template_string"),
call_sink("Template.render"),
call_sink("Environment.from_string"),
call_sink("jinja2.Environment.from_string"),
call_sink("mako.template.Template"),
],
sanitizers: vec![
call_sink("markupsafe.escape"),
call_sink("jinja2.escape"),
call_sink("html.escape"),
call_sink("cgi.escape"),
call_sink("bleach.clean"),
],
}
}
}
impl_rule! {
TaintSsti,
id = "py/taint-ssti",
severity = Severity::Critical,
cwe = Some("CWE-1336"),
description = "Untrusted input reaches template rendering sink (potential SSTI)",
language = Language::Python,
fn check_with_context(_self, source, tree, ctx) {
let meta = TaintRuleMeta {
rule_id: _self.id(),
severity: _self.severity(),
cwe: _self.cwe(),
fix_suggestion: Some(
"Use render_template() with separate template files instead of render_template_string() with user input",
),
};
map_taint_findings(&meta, source, tree, ctx, &Self::spec(), |src, sink| {
format!(
"{} reaches {} — untrusted input can inject server-side templates",
src, sink
)
})
}
}
pub struct TaintXpathInjection;
impl TaintXpathInjection {
fn spec() -> TaintSpec {
TaintSpec {
sources: python_taint_sources(),
sinks: vec![
call_sink("etree.XPath"),
call_sink("etree.xpath"),
call_sink("lxml.etree.XPath"),
NodeMatcher::MethodName {
method: "xpath".into(),
description: "lxml.etree.xpath".into(),
},
],
sanitizers: vec![],
}
}
}
impl_rule! {
TaintXpathInjection,
id = "py/taint-xpath-injection",
severity = Severity::High,
cwe = Some("CWE-643"),
description = "Untrusted input reaches XPath query sink",
language = Language::Python,
fn check_with_context(_self, source, tree, ctx) {
let meta = TaintRuleMeta {
rule_id: _self.id(),
severity: _self.severity(),
cwe: _self.cwe(),
fix_suggestion: Some(
"Use parameterized XPath queries or validate/sanitize input before building XPath expressions",
),
};
map_taint_findings(&meta, source, tree, ctx, &Self::spec(), |src, sink| {
format!(
"{} reaches {} — untrusted input can inject XPath expressions",
src, sink
)
})
}
}
pub struct TaintLdapInjection;
impl TaintLdapInjection {
fn spec() -> TaintSpec {
TaintSpec {
sources: python_taint_sources(),
sinks: vec![
call_sink("ldap.search_s"),
call_sink("ldap.search_st"),
call_sink("ldap.search_ext_s"),
call_sink("ldap3.Connection.search"),
call_sink("conn.search_s"),
call_sink("conn.search_st"),
call_sink("conn.search_ext_s"),
call_sink("l.search_s"),
call_sink("l.search_st"),
call_sink("l.search_ext_s"),
],
sanitizers: vec![
call_sink("ldap.filter.escape_filter_chars"),
call_sink("ldap3.utils.conv.escape_filter_chars"),
],
}
}
}
impl_rule! {
TaintLdapInjection,
id = "py/taint-ldap-injection",
severity = Severity::High,
cwe = Some("CWE-90"),
description = "Untrusted input reaches LDAP search sink",
language = Language::Python,
fn check_with_context(_self, source, tree, ctx) {
let meta = TaintRuleMeta {
rule_id: _self.id(),
severity: _self.severity(),
cwe: _self.cwe(),
fix_suggestion: Some(
"Use ldap.filter.escape_filter_chars() to sanitize user input before building LDAP filters",
),
};
map_taint_findings(&meta, source, tree, ctx, &Self::spec(), |src, sink| {
format!(
"{} reaches {} — untrusted input can inject LDAP filters",
src, sink
)
})
}
}
pub struct TaintLogInjection;
impl TaintLogInjection {
fn spec() -> TaintSpec {
TaintSpec {
sources: python_taint_sources(),
sinks: vec![
call_sink("logging.info"),
call_sink("logging.warning"),
call_sink("logging.error"),
call_sink("logging.debug"),
call_sink("logging.critical"),
call_sink("logging.log"),
call_sink("print"),
NodeMatcher::MethodName {
method: "info".into(),
description: "logger.info".into(),
},
NodeMatcher::MethodName {
method: "warning".into(),
description: "logger.warning".into(),
},
NodeMatcher::MethodName {
method: "error".into(),
description: "logger.error".into(),
},
NodeMatcher::MethodName {
method: "debug".into(),
description: "logger.debug".into(),
},
NodeMatcher::MethodName {
method: "critical".into(),
description: "logger.critical".into(),
},
],
sanitizers: vec![],
}
}
}
impl_rule! {
TaintLogInjection,
id = "py/taint-log-injection",
severity = Severity::Medium,
cwe = Some("CWE-117"),
description = "Untrusted input reaches a logging sink — possible log injection",
language = Language::Python,
fn check_with_context(_self, source, tree, ctx) {
let meta = TaintRuleMeta {
rule_id: _self.id(),
severity: _self.severity(),
cwe: _self.cwe(),
fix_suggestion: Some(
"Sanitize user input before logging — strip newlines and control characters (no standard sanitizer; use str.replace() or a regex to remove \\n, \\r, and ANSI escape sequences)",
),
};
map_taint_findings(&meta, source, tree, ctx, &Self::spec(), |src, sink| {
format!(
"{} reaches {} — untrusted input can forge log entries",
src, sink
)
})
}
}
pub struct TaintXxe;
impl TaintXxe {
fn spec() -> TaintSpec {
TaintSpec {
sources: python_taint_sources(),
sinks: vec![
call_sink("etree.parse"),
call_sink("etree.fromstring"),
call_sink("xml.etree.ElementTree.parse"),
call_sink("xml.etree.ElementTree.fromstring"),
call_sink("ElementTree.parse"),
call_sink("ElementTree.fromstring"),
call_sink("xml.sax.parseString"),
call_sink("sax.parseString"),
call_sink("minidom.parseString"),
call_sink("xml.dom.minidom.parseString"),
call_sink("pulldom.parse"),
call_sink("xml.dom.pulldom.parse"),
],
sanitizers: vec![
call_sink("defusedxml.parse"),
call_sink("defusedxml.fromstring"),
call_sink("defusedxml.ElementTree.parse"),
call_sink("defusedxml.minidom.parseString"),
],
}
}
}
impl_rule! {
TaintXxe,
id = "py/taint-xxe",
severity = Severity::High,
cwe = Some("CWE-611"),
description = "Untrusted input reaches an XML parser — possible XML External Entity (XXE) injection",
language = Language::Python,
fn check_with_context(_self, source, tree, ctx) {
let meta = TaintRuleMeta {
rule_id: _self.id(),
severity: _self.severity(),
cwe: _self.cwe(),
fix_suggestion: Some(
"Use defusedxml instead of xml.etree.ElementTree for untrusted XML input",
),
};
map_taint_findings(&meta, source, tree, ctx, &Self::spec(), |src, sink| {
format!(
"{} reaches {} — untrusted input can trigger XML External Entity processing",
src, sink
)
})
}
}
pub struct TaintNosqlInjection;
impl TaintNosqlInjection {
fn spec() -> TaintSpec {
TaintSpec {
sources: python_taint_sources(),
sinks: vec![
NodeMatcher::Call {
canonical: "collection.find".into(),
description: "MongoDB collection.find()".into(),
},
NodeMatcher::Call {
canonical: "db.find".into(),
description: "MongoDB db.find()".into(),
},
NodeMatcher::MethodName {
method: "find_one".into(),
description: "collection.find_one".into(),
},
NodeMatcher::MethodName {
method: "update_one".into(),
description: "collection.update_one".into(),
},
NodeMatcher::MethodName {
method: "update_many".into(),
description: "collection.update_many".into(),
},
NodeMatcher::MethodName {
method: "delete_one".into(),
description: "collection.delete_one".into(),
},
NodeMatcher::MethodName {
method: "delete_many".into(),
description: "collection.delete_many".into(),
},
NodeMatcher::MethodName {
method: "aggregate".into(),
description: "collection.aggregate".into(),
},
NodeMatcher::MethodName {
method: "count_documents".into(),
description: "collection.count_documents".into(),
},
],
sanitizers: vec![],
}
}
}
impl_rule! {
TaintNosqlInjection,
id = "py/taint-nosql-injection",
severity = Severity::High,
cwe = Some("CWE-943"),
description = "Untrusted input reaches a MongoDB query sink — possible NoSQL injection",
language = Language::Python,
fn check_with_context(_self, source, tree, ctx) {
let meta = TaintRuleMeta {
rule_id: _self.id(),
severity: _self.severity(),
cwe: _self.cwe(),
fix_suggestion: Some("Validate and sanitize user input before using in MongoDB queries. Avoid passing raw user input as query filters."),
};
map_taint_findings(&meta, source, tree, ctx, &Self::spec(), |src, sink| {
format!(
"{} reaches {} — untrusted input can inject NoSQL operators",
src, sink
)
})
}
}
pub fn python_taint_rule_specs() -> Vec<(&'static str, TaintSpec)> {
vec![
(
"py/taint-pickle-deserialization",
TaintPickleDeserialization::spec(),
),
("py/taint-eval", TaintEvalFromRequest::spec()),
(
"py/taint-command-injection",
TaintCommandInjectionFromRequest::spec(),
),
("py/taint-ssrf", TaintSsrfFromRequest::spec()),
("py/taint-yaml-load", TaintYamlLoadFromRequest::spec()),
(
"py/taint-sql-injection",
TaintSqlInjectionFromRequest::spec(),
),
("py/taint-ssti", TaintSsti::spec()),
("py/taint-xpath-injection", TaintXpathInjection::spec()),
("py/taint-ldap-injection", TaintLdapInjection::spec()),
("py/taint-log-injection", TaintLogInjection::spec()),
("py/taint-xxe", TaintXxe::spec()),
("py/taint-nosql-injection", TaintNosqlInjection::spec()),
]
}