use regex::Regex;
use std::sync::LazyLock;
use tracing::debug;
use crate::engine::Severity;
use crate::engine::diff_parser::{DiffLineType, FileChunk};
use crate::engine::rules::builtin::post_match_filter;
use crate::engine::rules::types::RuleFinding;
pub struct SecurityPattern {
pub id: &'static str,
pub name: &'static str,
pub regex: &'static str,
pub severity: Severity,
}
pub static PATTERNS: &[SecurityPattern] = &[
SecurityPattern {
id: "crypto/md5-password",
name: "MD5 used for password hashing",
regex: r"(?i)md5\s*\(\s*(?:password|passwd|pwd|secret)",
severity: Severity::Critical,
},
SecurityPattern {
id: "crypto/sha1-password",
name: "SHA-1 used for password hashing",
regex: r"(?i)sha1\s*\(\s*(?:password|passwd|pwd|secret)",
severity: Severity::Major,
},
SecurityPattern {
id: "crypto/weak-hash",
name: "Weak hash algorithm (MD5/SHA1) for security-sensitive data",
regex: r"(?i)(?:hashlib\.md5|hashlib\.sha1|MD5Create|SHA1Create|Digest::MD5|Digest::SHA1)",
severity: Severity::Major,
},
SecurityPattern {
id: "crypto/hardcoded-secret",
name: "Hardcoded password or secret in variable",
regex: r#"(?i)(?:password|passwd|pwd|secret|api_key|apikey|token)["']?(?:\s*:\s*[&\w<>\[\].?|]+)?\s*(?:[=:]|:=\s*["'`])\s*\S{8,}|(?:password|passwd|pwd|secret|api_key|apikey|token)["']?\s+[&*\w<>\[\].?|]+\s*=\s*["'`]\S{7,}|\b(?:password|identified\s+by)\s+'[^']+'"#,
severity: Severity::Critical,
},
SecurityPattern {
id: "injection/sql-concat",
name: "SQL injection via string concatenation",
regex: r#"(?i)(?:(?:SELECT|INSERT|UPDATE|DELETE)\b[^"\']*["\x27`][^"\']*\+|format!\s*\(\s*["\x27`]\s*(?:SELECT|INSERT|UPDATE|DELETE))"#,
severity: Severity::Critical,
},
SecurityPattern {
id: "injection/eval",
name: "eval() with dynamic input",
regex: r"(?i)eval\s*\(\s*(?:req|request|input|params|data|user)",
severity: Severity::Critical,
},
SecurityPattern {
id: "injection/exec",
name: "Command injection via exec/system with dynamic input",
regex: r#"(?i)(?:exec|system|popen|subprocess\.(?:call|run))\s*\((?:[^)]*(?:f"|f'|format\(|\.format\(|shell\s*=\s*True|\+\s*(?:req|request|input|params|data|user|query)|%s|%\(.*\)))"#,
severity: Severity::Critical,
},
SecurityPattern {
id: "auth/hardcoded-role",
name: "Hardcoded role or permission check",
regex: r#"(?i)(?:\w+\.)*role\s*[=]{2,3}\s*["']?(?:admin|super|root|superuser)["']?|is_admin\s*==\s*True"#,
severity: Severity::Major,
},
SecurityPattern {
id: "config/debug-enabled",
name: "Debug mode enabled (production risk)",
regex: r#"(?i)(?:DEBUG\s*=\s*True|debug\s*:\s*true|debug\s*=\s*true)"#,
severity: Severity::Minor,
},
SecurityPattern {
id: "config/cors-wildcard",
name: "CORS wildcard allows all origins",
regex: r#"(?ix)
(?:
# 1. HTTP response header (any spacing/quote style)
Access-Control-Allow-Origin \s* :? \s* ["']? \* ["']? (?: \s | ; | $ )
|
# 2. Generic keyword assignment with wildcard (covers quotes, brackets)
(?: cors | allow_origin | allowed_origins | allow_origins ) \s* [=:(\[] \s* ["'\[\{]{0,2} \* ["'\]\}]{0,2} (?: \s | ; | , | \) | \] | $ )
|
# 3. keyword origin assignment
origin \s* [:=] \s* ["']? \* ["']? (?: \s | ; | $ )
|
# 4. Django boolean flag
CORS_ALLOW_ALL_ORIGINS \s* = \s* True
|
# 5. Spring .allowedOrigins("*")
\. allowedOrigins \s* \( \s* ["'] \* ["'] \s* \)
|
# 6. .NET AllowAnyOrigin()
AllowAnyOrigin \s* \(
|
# 7. tower-http allow_origin(Any)
allow_origin \s* \( \s* Any \s* \)
|
# 8. Express cors({ origin: true })
cors \s* \( \s* \{ \s* origin \s* : \s* true
|
# 9. nginx add_header
add_header \s+ Access-Control-Allow-Origin \s+ \*
|
# 10. actix SetHeader
SetHeader \s* \( \s* ["'] Access-Control-Allow-Origin ["'] \s* , \s* ["'] \* ["']
)"#,
severity: Severity::Major,
},
SecurityPattern {
id: "crypto/ssl-verify-disabled",
name: "SSL certificate verification disabled",
regex: r"(?i)(?:verify\s*=\s*False|verify:\s*false|rejectUnauthorized:\s*false)",
severity: Severity::Critical,
},
];
static COMPILED_PATTERNS: LazyLock<Vec<(String, String, Regex, Severity)>> = LazyLock::new(|| {
PATTERNS
.iter()
.filter_map(|p| {
Regex::new(p.regex)
.ok()
.map(|re| (p.id.to_string(), p.name.to_string(), re, p.severity))
})
.collect()
});
pub fn scan_security(chunks: &[FileChunk], max_findings: usize) -> Vec<RuleFinding> {
let mut findings = Vec::new();
for chunk in chunks {
let path = chunk
.new_path
.as_deref()
.or(chunk.old_path.as_deref())
.unwrap_or("unknown");
if is_test_file(path) || is_doc_file(path) {
debug!(
file = path,
"test/doc file: only high-confidence secret scan"
);
scan_high_confidence_secrets(chunk, path, &mut findings, max_findings);
if findings.len() >= max_findings {
findings.sort_by_key(|f| std::cmp::Reverse(f.severity));
return findings;
}
continue;
}
for hunk in &chunk.chunks {
for line in &hunk.lines {
if line.line_type != DiffLineType::Add {
continue;
}
let line_no = line.new_line_no.unwrap_or(0);
if line_no == 0 {
continue;
}
for (rule_id, name, regex, severity) in COMPILED_PATTERNS.iter() {
if regex.is_match(&line.content) {
if post_match_filter(rule_id, &line.content) {
debug!(
rule = %rule_id,
file = path,
line = line_no,
"security pattern suppressed by post-match filter"
);
continue;
}
debug!(
rule = %rule_id,
file = path,
line = line_no,
"security pattern match"
);
findings.push(RuleFinding {
rule_id: rule_id.clone(),
file: path.to_string(),
line: line_no,
severity: *severity,
title: name.clone(),
body: format!(
"Static security scanner detected: {} in {}:{}",
name, path, line_no
),
});
if findings.len() >= max_findings {
findings.sort_by_key(|f| std::cmp::Reverse(f.severity));
return findings;
}
}
}
}
}
}
findings.sort_by_key(|f| std::cmp::Reverse(f.severity));
findings
}
fn scan_high_confidence_secrets(
chunk: &FileChunk,
path: &str,
findings: &mut Vec<RuleFinding>,
max_findings: usize,
) {
for hunk in &chunk.chunks {
for line in &hunk.lines {
if line.line_type != DiffLineType::Add {
continue;
}
let line_no = line.new_line_no.unwrap_or(0);
if line_no == 0 {
continue;
}
if crate::engine::secret_patterns::find_high_confidence(&line.content).is_none() {
continue;
}
findings.push(RuleFinding {
rule_id: "secrets/high-confidence-in-test-or-doc".to_string(),
file: path.to_string(),
line: line_no,
severity: Severity::Major,
title: "Credential-like secret in test/doc file".to_string(),
body: format!(
"Static security scanner detected a credential-like value in {path}:{line_no}. \
Test/doc paths are not exempt from secret scanning; verify it is a fake."
),
});
if findings.len() >= max_findings {
return;
}
}
}
}
fn is_test_file(path: &str) -> bool {
let lower = path.to_lowercase();
for seg in lower.split(['/', '_', '-', '.']) {
if matches!(
seg,
"test"
| "tests"
| "testing"
| "tested"
| "__tests__"
| "spec"
| "specs"
| "fixture"
| "fixtures"
| "mock"
| "mocks"
| "example"
| "examples"
) {
return true;
}
}
false
}
fn is_doc_file(path: &str) -> bool {
let lower = path.to_lowercase();
let ext = match lower.rfind('.') {
Some(pos) => &lower[pos + 1..],
None => return false,
};
matches!(
ext,
"md" | "markdown" | "mdx" | "txt" | "rst" | "adoc" | "asciidoc" | "tex" | "org"
)
}
#[cfg(test)]
mod tests {
use super::*;
fn make_chunk(file: &str, added_lines: &[&str]) -> FileChunk {
use crate::engine::diff_parser::{DiffHunk, DiffLine};
FileChunk {
old_path: None,
new_path: Some(file.to_string()),
language: "py".to_string(),
chunks: vec![DiffHunk {
old_start: 1,
old_count: 1,
new_start: 1,
new_count: added_lines.len() as u32,
header: String::new(),
lines: added_lines
.iter()
.enumerate()
.map(|(i, content)| DiffLine {
line_type: DiffLineType::Add,
content: content.to_string(),
old_line_no: None,
new_line_no: Some(i as u32 + 1),
})
.collect(),
}],
is_binary: false,
is_deleted: false,
is_new: false,
}
}
#[test]
fn typed_declarations_with_literal_secrets_are_flagged() {
for line in [
r#"const password: string = "hunter2hunter2xx";"#,
r#"password: str = "hunter2hunter2xx""#,
r#"let password: &str = "hunter2hunter2xx";"#,
r#"private val apiSecret: String = "hunter2hunter2xx""#,
] {
let found = scan_security(&[make_chunk("src/app.ts", &[line])], 50);
assert!(
found.iter().any(|f| f.line == 1),
"typed secret not flagged: {line}"
);
}
}
#[test]
fn typed_declaration_without_literal_is_not_flagged() {
for line in [
"const password: string = process.env.PASSWORD_VALUE;",
"password: str",
] {
let found = scan_security(&[make_chunk("src/app.ts", &[line])], 50);
assert!(
!found.iter().any(|f| f.title.contains("Hardcoded")),
"false positive: {line} -> {:?}",
found.iter().map(|f| f.title.clone()).collect::<Vec<_>>()
);
}
}
#[test]
fn test_and_doc_paths_still_scan_high_confidence_secrets() {
let key = format!("{}{}", "AKIA", "QWERTYUIOPASDFGH");
let line = format!("key = {key}");
for path in [
"tests/x.rs",
"src/__tests__/a.ts",
"foo.md",
"docs/notes.txt",
] {
let chunks = vec![make_chunk(path, &[line.as_str()])];
let findings = scan_security(&chunks, 10);
assert_eq!(findings.len(), 1, "{path} should be scanned for secrets");
assert_eq!(
findings[0].rule_id,
"secrets/high-confidence-in-test-or-doc"
);
}
let pk = vec![make_chunk(
"tests/k.rs",
&["-----BEGIN RSA PRIVATE KEY-----"],
)];
assert_eq!(scan_security(&pk, 10).len(), 1);
}
#[test]
fn test_and_doc_paths_stay_quiet_for_noisy_rules_and_placeholders() {
let chunks = vec![
make_chunk("tests/auth.rs", &["let password = supersecret123;"]),
make_chunk("README.md", &["Set CORS to * and debug = true"]),
make_chunk("tests/aws.rs", &["AKIAIOSFODNN7EXAMPLE"]),
];
assert!(scan_security(&chunks, 10).is_empty());
}
#[test]
fn detects_hardcoded_password() {
let chunks = vec![make_chunk(
"src/auth.rs",
&["let password = supersecret123;"],
)];
let findings = scan_security(&chunks, 10);
assert!(!findings.is_empty());
assert!(findings[0].rule_id.contains("hardcoded-secret"));
}
#[test]
fn detects_sql_injection() {
let chunks = vec![make_chunk(
"src/db.py",
&["query = \"SELECT * FROM users WHERE id = \" + req.params.id"],
)];
let findings = scan_security(&chunks, 10);
assert!(!findings.is_empty());
assert!(findings[0].rule_id.contains("sql"));
}
#[test]
fn detects_eval_injection() {
let chunks = vec![make_chunk("src/run.js", &["eval(request.body.code)"])];
let findings = scan_security(&chunks, 10);
assert!(!findings.is_empty());
assert!(findings[0].rule_id.contains("eval"));
}
#[test]
fn detects_debug_enabled() {
let chunks = vec![make_chunk("src/config.py", &["DEBUG = True"])];
let findings = scan_security(&chunks, 10);
assert!(!findings.is_empty());
assert!(findings[0].rule_id.contains("debug"));
}
#[test]
fn detects_ssl_verify_disabled() {
let chunks = vec![make_chunk(
"src/client.py",
&["requests.get(url, verify=False)"],
)];
let findings = scan_security(&chunks, 10);
assert!(!findings.is_empty());
assert!(findings[0].rule_id.contains("ssl"));
}
#[test]
fn detects_cors_wildcard() {
let chunks = vec![make_chunk(
"src/server.rs",
&["Access-Control-Allow-Origin: *"],
)];
let findings = scan_security(&chunks, 10);
assert!(!findings.is_empty());
}
#[test]
fn skips_test_files() {
let chunks = vec![make_chunk(
"tests/auth_test.py",
&["let password = test_password_123;"],
)];
let findings = scan_security(&chunks, 10);
assert!(findings.is_empty());
}
#[test]
fn is_test_file_does_not_over_match_common_words() {
assert!(!is_test_file("src/latest_config.rs"));
assert!(!is_test_file("src/models/attestation.rs"));
assert!(!is_test_file("src/utils/aspect.rs"));
assert!(!is_test_file("src/protest.rs"));
assert!(!is_test_file("src/inspector.rs"));
assert!(is_test_file("tests/auth_test.py"));
assert!(is_test_file("src/app.test.ts"));
assert!(is_test_file("src/__tests__/setup.rs"));
assert!(is_test_file("spec/models/user_spec.rb"));
}
#[test]
fn empty_diff_no_findings() {
let findings = scan_security(&[], 10);
assert!(findings.is_empty());
}
#[test]
fn respects_max_findings() {
let chunks = vec![
make_chunk("src/a.py", &["DEBUG = True", "eval(input.data)"]),
make_chunk("src/b.py", &["let password = supersecret"]),
];
let findings = scan_security(&chunks, 2);
assert_eq!(findings.len(), 2);
}
#[test]
fn sorts_by_severity() {
let chunks = vec![make_chunk(
"src/app.py",
&["DEBUG = True", "let password = supersecret"],
)];
let findings = scan_security(&chunks, 10);
assert!(findings[0].severity >= findings[findings.len() - 1].severity);
}
#[test]
fn subprocess_run_with_literal_list_no_false_positive() {
let chunks = vec![make_chunk(
"src/provider.py",
&["cmd = [self._bin, 'recall', query]", "subprocess.run(cmd)"],
)];
let findings = scan_security(&chunks, 10);
let exec_findings: Vec<_> = findings
.iter()
.filter(|f| f.rule_id == "injection/exec")
.collect();
assert!(
exec_findings.is_empty(),
"subprocess.run(cmd) with controlled list should not trigger"
);
}
#[test]
fn subprocess_run_with_fstring_triggers() {
let chunks = vec![make_chunk(
"src/handler.py",
&["subprocess.run(f'cat {user_input}')"],
)];
let findings = scan_security(&chunks, 10);
let exec_findings: Vec<_> = findings
.iter()
.filter(|f| f.rule_id == "injection/exec")
.collect();
assert_eq!(
exec_findings.len(),
1,
"subprocess.run with f-string should trigger"
);
}
#[test]
fn subprocess_run_with_shell_true_triggers() {
let chunks = vec![make_chunk(
"src/handler.py",
&["subprocess.run(cmd, shell=True)"],
)];
let findings = scan_security(&chunks, 10);
let exec_findings: Vec<_> = findings
.iter()
.filter(|f| f.rule_id == "injection/exec")
.collect();
assert_eq!(
exec_findings.len(),
1,
"subprocess.run with shell=True should trigger"
);
}
#[test]
fn svg_xmlns_url_suppressed_in_security_scan() {
let chunks = vec![make_chunk(
"src/icon.svg",
&[r#"<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24">"#],
)];
let findings = scan_security(&chunks, 10);
assert!(
findings.is_empty(),
"SVG xmlns should not trigger security findings"
);
}
#[test]
fn empty_secret_value_suppressed_in_security_scan() {
let chunks = vec![make_chunk(
"src/Form.svelte",
&["let formAppSecret = $state('');"],
)];
let findings = scan_security(&chunks, 10);
let secret_findings: Vec<_> = findings
.iter()
.filter(|f| f.rule_id == "crypto/hardcoded-secret")
.collect();
assert!(
secret_findings.is_empty(),
"Empty $state('') secret should not trigger"
);
}
#[test]
fn real_hardcoded_secret_still_detected_after_filter() {
let chunks = vec![make_chunk(
"src/config.py",
&["API_KEY = \"sk_live_abc123def456ghi789\""],
)];
let findings = scan_security(&chunks, 10);
let secret_findings: Vec<_> = findings
.iter()
.filter(|f| f.rule_id == "crypto/hardcoded-secret")
.collect();
assert_eq!(
secret_findings.len(),
1,
"Real hardcoded secret should still be detected"
);
}
#[test]
fn detects_cors_wildcard_header() {
let chunks = vec![make_chunk(
"src/server.rs",
&["Access-Control-Allow-Origin: *"],
)];
let findings = scan_security(&chunks, 10);
let cors_findings: Vec<_> = findings
.iter()
.filter(|f| f.rule_id == "config/cors-wildcard")
.collect();
assert_eq!(cors_findings.len(), 1, "Should detect wildcard CORS header");
}
#[test]
fn detects_cors_wildcard_assignment() {
let chunks = vec![make_chunk("src/config.rs", &["let cors_origin = \"*\";"])];
let findings = scan_security(&chunks, 10);
let cors_findings: Vec<_> = findings
.iter()
.filter(|f| f.rule_id == "config/cors-wildcard")
.collect();
assert_eq!(
cors_findings.len(),
1,
"Should detect cors assignment with wildcard"
);
}
#[test]
fn detects_allowed_origins_wildcard() {
let chunks = vec![make_chunk("src/app.py", &["allowed_origins = \"*\""])];
let findings = scan_security(&chunks, 10);
let cors_findings: Vec<_> = findings
.iter()
.filter(|f| f.rule_id == "config/cors-wildcard")
.collect();
assert_eq!(
cors_findings.len(),
1,
"Should detect allowed_origins with wildcard"
);
}
#[test]
fn no_false_positive_cors_env_var_name() {
let chunks = vec![make_chunk(
"src/config.rs",
&["let val = std::env::var(\"TITEN_CORS_ORIGINS\").unwrap_or(\"*\");"],
)];
let findings = scan_security(&chunks, 10);
let cors_findings: Vec<_> = findings
.iter()
.filter(|f| f.rule_id == "config/cors-wildcard")
.collect();
assert!(
cors_findings.is_empty(),
"TITEN_CORS_ORIGINS env var name should not trigger CORS wildcard"
);
}
#[test]
fn no_false_positive_cors_in_prose() {
let chunks = vec![make_chunk(
"src/config.rs",
&["// CORS configured via TITEN_CORS_ORIGINS env var"],
)];
let findings = scan_security(&chunks, 10);
let cors_findings: Vec<_> = findings
.iter()
.filter(|f| f.rule_id == "config/cors-wildcard")
.collect();
assert!(
cors_findings.is_empty(),
"CORS keyword in comment prose should not trigger"
);
}
#[test]
fn no_false_positive_markdown_file() {
let chunks = vec![make_chunk(
"docs/deployment.md",
&["Access-Control-Allow-Origin: *"],
)];
let findings = scan_security(&chunks, 10);
assert!(
findings.is_empty(),
"Markdown files should not be scanned by security scanner"
);
}
#[test]
fn no_false_positive_txt_file() {
let chunks = vec![make_chunk(
"docs/security.txt",
&["Access-Control-Allow-Origin: *"],
)];
let findings = scan_security(&chunks, 10);
assert!(
findings.is_empty(),
"Text files should not be scanned by security scanner"
);
}
#[test]
fn no_false_positive_rst_file() {
let chunks = vec![make_chunk(
"docs/api.rst",
&["Access-Control-Allow-Origin: *"],
)];
let findings = scan_security(&chunks, 10);
assert!(
findings.is_empty(),
"reStructuredText files should not be scanned"
);
}
#[test]
fn real_cors_wildcard_in_rust_still_detected() {
let chunks = vec![make_chunk(
"src/server.rs",
&[".layer(CorsLayer::new().allow_origin(\"*\"))"],
)];
let findings = scan_security(&chunks, 10);
let cors_findings: Vec<_> = findings
.iter()
.filter(|f| f.rule_id == "config/cors-wildcard")
.collect();
assert_eq!(
cors_findings.len(),
1,
"Real CORS wildcard in Rust code should be detected"
);
}
#[test]
fn detects_cors_wildcard_nginx_add_header() {
let chunks = vec![make_chunk(
"nginx.conf",
&["add_header Access-Control-Allow-Origin *;"],
)];
let findings = scan_security(&chunks, 10);
let cors: Vec<_> = findings
.iter()
.filter(|f| f.rule_id == "config/cors-wildcard")
.collect();
assert_eq!(
cors.len(),
1,
"nginx add_header wildcard should be detected"
);
}
#[test]
fn detects_cors_wildcard_django() {
let chunks = vec![make_chunk(
"settings.py",
&["CORS_ALLOW_ALL_ORIGINS = True"],
)];
let findings = scan_security(&chunks, 10);
let cors: Vec<_> = findings
.iter()
.filter(|f| f.rule_id == "config/cors-wildcard")
.collect();
assert_eq!(
cors.len(),
1,
"Django CORS_ALLOW_ALL_ORIGINS should be detected"
);
}
#[test]
fn detects_cors_wildcard_spring_boot() {
let chunks = vec![make_chunk("WebConfig.java", &[".allowedOrigins(\"*\")"])];
let findings = scan_security(&chunks, 10);
let cors: Vec<_> = findings
.iter()
.filter(|f| f.rule_id == "config/cors-wildcard")
.collect();
assert_eq!(
cors.len(),
1,
"Spring .allowedOrigins(\"*\") should be detected"
);
}
#[test]
fn detects_cors_wildcard_dotnet_allowany() {
let chunks = vec![make_chunk("Startup.cs", &[".AllowAnyOrigin()"])];
let findings = scan_security(&chunks, 10);
let cors: Vec<_> = findings
.iter()
.filter(|f| f.rule_id == "config/cors-wildcard")
.collect();
assert_eq!(cors.len(), 1, ".NET AllowAnyOrigin() should be detected");
}
#[test]
fn detects_cors_wildcard_tower_http_any() {
let chunks = vec![make_chunk("src/main.rs", &[".allow_origin(Any)"])];
let findings = scan_security(&chunks, 10);
let cors: Vec<_> = findings
.iter()
.filter(|f| f.rule_id == "config/cors-wildcard")
.collect();
assert_eq!(
cors.len(),
1,
"tower-http allow_origin(Any) should be detected"
);
}
#[test]
fn detects_cors_wildcard_express_origin_true() {
let chunks = vec![make_chunk("app.js", &["cors({ origin: true })"])];
let findings = scan_security(&chunks, 10);
let cors: Vec<_> = findings
.iter()
.filter(|f| f.rule_id == "config/cors-wildcard")
.collect();
assert_eq!(
cors.len(),
1,
"Express cors({{ origin: true }}) should be detected"
);
}
#[test]
fn detects_cors_wildcard_fastapi_list() {
let chunks = vec![make_chunk("main.py", &["allow_origins=[\"*\"]"])];
let findings = scan_security(&chunks, 10);
let cors: Vec<_> = findings
.iter()
.filter(|f| f.rule_id == "config/cors-wildcard")
.collect();
assert_eq!(
cors.len(),
1,
"FastAPI allow_origins=[\"*\"] should be detected"
);
}
#[test]
fn detects_cors_wildcard_nginx_set_header_actix() {
let chunks = vec![make_chunk(
"src/server.rs",
&["SetHeader(\"Access-Control-Allow-Origin\", \"*\")"],
)];
let findings = scan_security(&chunks, 10);
let cors: Vec<_> = findings
.iter()
.filter(|f| f.rule_id == "config/cors-wildcard")
.collect();
assert_eq!(cors.len(), 1, "actix SetHeader wildcard should be detected");
}
#[test]
fn is_doc_file_does_not_match_extensionless() {
assert!(!is_doc_file("org"));
assert!(!is_doc_file("tex"));
assert!(!is_doc_file("md"));
assert!(!is_doc_file("src/org"));
assert!(!is_doc_file("bin/tex"));
}
#[test]
fn is_doc_file_recognizes_common_extensions() {
assert!(is_doc_file("README.md"));
assert!(is_doc_file("docs/guide.markdown"));
assert!(is_doc_file("docs/api.mdx"));
assert!(is_doc_file("notes.txt"));
assert!(is_doc_file("docs/spec.rst"));
assert!(is_doc_file("docs/manual.adoc"));
assert!(is_doc_file("docs/manual.asciidoc"));
assert!(is_doc_file("paper.tex"));
assert!(is_doc_file("notes.org"));
}
#[test]
fn is_doc_file_does_not_match_code_files() {
assert!(!is_doc_file("src/main.rs"));
assert!(!is_doc_file("src/app.py"));
assert!(!is_doc_file("src/server.ts"));
assert!(!is_doc_file("src/index.js"));
assert!(!is_doc_file("src/config.go"));
assert!(!is_doc_file("Dockerfile"));
assert!(!is_doc_file("docker-compose.yml"));
assert!(!is_doc_file("Makefile"));
}
#[test]
fn sql_concat_real_injection_still_detected() {
let chunks = vec![make_chunk(
"src/db.py",
&["query = \"SELECT * FROM users WHERE id = \" + user_input"],
)];
let findings = scan_security(&chunks, 10);
assert!(!findings.is_empty(), "real SQL concat must be detected");
assert!(findings[0].rule_id.contains("sql"));
}
#[test]
fn sql_concat_format_macro_detected() {
let chunks = vec![make_chunk(
"src/db.rs",
&["let q = format!(\"SELECT * FROM users WHERE id = {}\", id);"],
)];
let findings = scan_security(&chunks, 10);
assert!(!findings.is_empty(), "format! with SELECT must be detected");
}
#[test]
fn sql_concat_comment_not_flagged() {
let chunks = vec![make_chunk(
"src/db.py",
&["// SELECT all users then concatenate results"],
)];
let findings = scan_security(&chunks, 10);
let sql_findings: Vec<_> = findings
.iter()
.filter(|f| f.rule_id.contains("sql"))
.collect();
assert!(
sql_findings.is_empty(),
"comment with SQL keyword + '+' should not be flagged"
);
}
#[test]
fn sql_concat_python_comment_not_flagged() {
let chunks = vec![make_chunk("src/db.py", &["# SELECT a + b FROM joined"])];
let findings = scan_security(&chunks, 10);
let sql_findings: Vec<_> = findings
.iter()
.filter(|f| f.rule_id.contains("sql"))
.collect();
assert!(sql_findings.is_empty());
}
#[test]
fn sql_concat_non_sql_plus_not_flagged() {
let chunks = vec![make_chunk(
"src/calc.rs",
&["let total = a + b; // UPDATE: not SQL"],
)];
let findings = scan_security(&chunks, 10);
let sql_findings: Vec<_> = findings
.iter()
.filter(|f| f.rule_id.contains("sql"))
.collect();
assert!(sql_findings.is_empty());
}
#[test]
fn debug_real_assignment_still_detected() {
let chunks = vec![make_chunk("src/config.py", &["DEBUG = True"])];
let findings = scan_security(&chunks, 10);
assert!(!findings.is_empty(), "DEBUG = True must be detected");
assert!(findings[0].rule_id.contains("debug"));
}
#[test]
fn debug_yaml_true_still_detected() {
let chunks = vec![make_chunk("src/config.yml", &["debug: true"])];
let findings = scan_security(&chunks, 10);
assert!(!findings.is_empty(), "debug: true must be detected");
}
#[test]
fn debug_cli_flag_not_flagged() {
let chunks = vec![make_chunk(
"Dockerfile",
&["RUN cargo test -- --debug 2>/dev/null"],
)];
let findings = scan_security(&chunks, 10);
let debug_findings: Vec<_> = findings
.iter()
.filter(|f| f.rule_id.contains("debug"))
.collect();
assert!(
debug_findings.is_empty(),
"--debug CLI flag should not be flagged"
);
}
#[test]
fn debug_argument_parser_not_flagged() {
let chunks = vec![make_chunk(
"src/cli.py",
&["parser.add_argument('--debug', help='Enable debug mode')"],
)];
let findings = scan_security(&chunks, 10);
let debug_findings: Vec<_> = findings
.iter()
.filter(|f| f.rule_id.contains("debug"))
.collect();
assert!(
debug_findings.is_empty(),
"argument parser definition should not be flagged"
);
}
#[test]
fn debug_comment_not_flagged() {
let chunks = vec![make_chunk(
"src/config.py",
&["# Use --debug for verbose output"],
)];
let findings = scan_security(&chunks, 10);
let debug_findings: Vec<_> = findings
.iter()
.filter(|f| f.rule_id.contains("debug"))
.collect();
assert!(debug_findings.is_empty());
}
#[test]
fn hardcoded_role_unquoted_still_detected() {
let chunks = vec![make_chunk("src/auth.rb", &["if role == admin"])];
let findings = scan_security(&chunks, 10);
assert!(!findings.is_empty(), "unquoted role check must be detected");
assert!(findings[0].rule_id.contains("role"));
}
#[test]
fn hardcoded_role_quoted_double_quotes_detected() {
let chunks = vec![make_chunk(
"src/auth.js",
&["if (role == \"admin\") return true;"],
)];
let findings = scan_security(&chunks, 10);
assert!(
!findings.is_empty(),
"quoted role == \"admin\" must be detected"
);
}
#[test]
fn hardcoded_role_quoted_single_quotes_detected() {
let chunks = vec![make_chunk("src/auth.py", &["if role == 'admin':"])];
let findings = scan_security(&chunks, 10);
assert!(
!findings.is_empty(),
"quoted role == 'admin' must be detected"
);
}
#[test]
fn hardcoded_role_strict_equality_detected() {
let chunks = vec![make_chunk(
"src/auth.ts",
&["if (role === \"admin\") return true;"],
)];
let findings = scan_security(&chunks, 10);
assert!(
!findings.is_empty(),
"strict equality role === must be detected"
);
}
#[test]
fn hardcoded_role_property_access_detected() {
let chunks = vec![make_chunk(
"src/auth.js",
&["if (user.role == \"admin\") return true;"],
)];
let findings = scan_security(&chunks, 10);
assert!(
!findings.is_empty(),
"user.role == \"admin\" must be detected"
);
}
#[test]
fn hardcoded_role_quoted_super_detected() {
let chunks = vec![make_chunk("src/auth.py", &["if role == \"super\":"])];
let findings = scan_security(&chunks, 10);
assert!(!findings.is_empty(), "role == \"super\" must be detected");
}
#[test]
fn hardcoded_role_quoted_root_detected() {
let chunks = vec![make_chunk("src/auth.py", &["if role == 'root':"])];
let findings = scan_security(&chunks, 10);
assert!(!findings.is_empty(), "role == 'root' must be detected");
}
#[test]
fn hardcoded_role_quoted_superuser_detected() {
let chunks = vec![make_chunk(
"src/auth.ts",
&["if (user.role === \"superuser\") grant_all();"],
)];
let findings = scan_security(&chunks, 10);
assert!(
!findings.is_empty(),
"role === \"superuser\" must be detected"
);
}
#[test]
fn eval_real_injection_still_detected() {
let chunks = vec![make_chunk("src/app.py", &["result = eval(request.data)"])];
let findings = scan_security(&chunks, 10);
assert!(!findings.is_empty(), "eval(request.data) must be detected");
}
#[test]
fn eval_user_input_still_detected() {
let chunks = vec![make_chunk("src/app.py", &["eval(user_input)"])];
let findings = scan_security(&chunks, 10);
assert!(!findings.is_empty(), "eval(user_input) must be detected");
}
#[test]
fn eval_comment_suppressed() {
let chunks = vec![make_chunk(
"src/app.py",
&["// eval(request.body) — deprecated, use safe_parse()"],
)];
let findings = scan_security(&chunks, 10);
assert!(findings.is_empty(), "eval in comment should be suppressed");
}
#[test]
fn evaluate_not_flagged_as_eval() {
let chunks = vec![make_chunk(
"src/app.py",
&["result = evaluate(request, context)"],
)];
let findings = scan_security(&chunks, 10);
assert!(findings.is_empty(), "evaluate() should not match eval rule");
}
#[test]
fn eval_docstring_suppressed() {
let chunks = vec![make_chunk(
"src/app.py",
&["\"\"\"Calls eval(params) for backwards compatibility.\"\"\""],
)];
let findings = scan_security(&chunks, 10);
assert!(
findings.is_empty(),
"eval in docstring should be suppressed"
);
}
#[test]
fn ast_literal_eval_suppressed() {
let chunks = vec![make_chunk(
"src/app.py",
&["data = ast.literal_eval(request.body)"],
)];
let findings = scan_security(&chunks, 10);
assert!(
findings.is_empty(),
"ast.literal_eval should be suppressed (safe)"
);
}
#[test]
fn weak_hash_real_usage_still_detected() {
let chunks = vec![make_chunk(
"src/crypto.py",
&["hashlib.md5(data).hexdigest()"],
)];
let findings = scan_security(&chunks, 10);
assert!(!findings.is_empty(), "hashlib.md5(data) must be detected");
}
#[test]
fn weak_hash_comment_suppressed() {
let chunks = vec![make_chunk(
"src/crypto.py",
&["// TODO: replace hashlib.md5 with sha256"],
)];
let findings = scan_security(&chunks, 10);
assert!(
findings.is_empty(),
"hashlib.md5 in comment should be suppressed"
);
}
#[test]
fn weak_hash_import_suppressed() {
let chunks = vec![make_chunk(
"src/crypto.py",
&["from hashlib import md5, sha1, sha256"],
)];
let findings = scan_security(&chunks, 10);
assert!(
findings.is_empty(),
"import of hashlib.md5 should be suppressed"
);
}
#[test]
fn weak_hash_rust_use_suppressed() {
let chunks = vec![make_chunk("src/crypto.rs", &["use sha1::Sha1;"])];
let findings = scan_security(&chunks, 10);
assert!(
findings.is_empty(),
"Rust use of Digest::SHA1 should be suppressed"
);
}
#[test]
fn weak_hash_docstring_suppressed() {
let chunks = vec![make_chunk(
"src/crypto.py",
&["\"\"\"Uses hashlib.md5 for legacy compatibility.\"\"\""],
)];
let findings = scan_security(&chunks, 10);
assert!(
findings.is_empty(),
"hashlib.md5 in docstring should be suppressed"
);
}
#[test]
fn ssl_verify_disabled_real_still_detected() {
let chunks = vec![make_chunk(
"src/client.py",
&["requests.get(url, verify=False)"],
)];
let findings = scan_security(&chunks, 10);
assert!(!findings.is_empty(), "verify=False must be detected");
}
#[test]
fn ssl_verify_disabled_reject_unauthorized_still_detected() {
let chunks = vec![make_chunk(
"src/client.ts",
&["agent: new https.Agent({ rejectUnauthorized: false })"],
)];
let findings = scan_security(&chunks, 10);
assert!(
!findings.is_empty(),
"rejectUnauthorized: false must be detected"
);
}
#[test]
fn ssl_verify_comment_suppressed() {
let chunks = vec![make_chunk(
"src/client.py",
&["# Do not set verify=False in production"],
)];
let findings = scan_security(&chunks, 10);
assert!(
findings.is_empty(),
"verify=False in comment should be suppressed"
);
}
#[test]
fn ssl_verify_env_reference_suppressed() {
let chunks = vec![make_chunk(
"src/client.py",
&["verify = os.environ.get('SSL_VERIFY', 'True')"],
)];
let findings = scan_security(&chunks, 10);
assert!(findings.is_empty(), "verify from env should be suppressed");
}
#[test]
fn ssl_verify_schema_definition_suppressed() {
let chunks = vec![make_chunk(
"src/config.ts",
&["interface HttpConfig { verify: false }"],
)];
let findings = scan_security(&chunks, 10);
assert!(
findings.is_empty(),
"verify: false in interface/schema should be suppressed"
);
}
#[test]
fn ssl_verify_true_not_flagged() {
let chunks = vec![make_chunk("src/client.py", &["verify: true"])];
let findings = scan_security(&chunks, 10);
assert!(findings.is_empty(), "verify: true should not be flagged");
}
#[test]
fn ssl_verify_docstring_suppressed() {
let chunks = vec![make_chunk(
"src/client.py",
&["\"\"\"Never use verify=False in production code.\"\"\""],
)];
let findings = scan_security(&chunks, 10);
assert!(
findings.is_empty(),
"verify=False in docstring should be suppressed"
);
}
}