diff --git a/src/pyspector/_rust_core/src/analysis/config_analysis.rs b/src/pyspector/_rust_core/src/analysis/config_analysis.rs index 88ee43e..a7ae0f6 100644 --- a/src/pyspector/_rust_core/src/analysis/config_analysis.rs +++ b/src/pyspector/_rust_core/src/analysis/config_analysis.rs @@ -1,5 +1,8 @@ use crate::issues::Issue; -use crate::rules::RuleSet; +use crate::rules::{EntropyRule, RuleSet}; +use regex::Regex; + +use super::entropy::shannon_entropy; pub fn scan_file(file_path: &str, content: &str, ruleset: &RuleSet) -> Vec { let mut issues = Vec::new(); @@ -32,24 +35,37 @@ pub fn scan_file(file_path: &str, content: &str, ruleset: &RuleSet) -> Vec Vec bool { let trimmed = line.trim(); - + // Skip obvious comments if trimmed.starts_with('#') { return true; } - + // Skip lines that are entirely string literals (docstrings) if (trimmed.starts_with("\"\"\"") && trimmed.ends_with("\"\"\"") && trimmed.len() > 6) || (trimmed.starts_with("'''") && trimmed.ends_with("'''") && trimmed.len() > 6) || @@ -78,14 +94,254 @@ fn is_in_comment_or_string(line: &str) -> bool { (trimmed.starts_with('\'') && trimmed.ends_with('\'') && !trimmed.contains(" = ")) { return true; } - + // More sophisticated check: if the line contains quotes but no assignment/function call // it's likely a standalone string/docstring - if (trimmed.contains("\"\"\"") || trimmed.contains("'''")) && - !trimmed.contains('=') && + if (trimmed.contains("\"\"\"") || trimmed.contains("'''")) && + !trimmed.contains('=') && !trimmed.contains('(') { return true; } - + false -} \ No newline at end of file +} + +/// Returns true if the given bytes look like binary content (contains a NUL +/// byte within the first 8KB) rather than text. Used to skip binary files +/// (images, archives, compiled artifacts) before pattern/entropy scanning. +pub fn looks_binary(bytes: &[u8]) -> bool { + let sample_len = bytes.len().min(8192); + bytes[..sample_len].contains(&0u8) +} + +/// Replaces the `[start, end)` byte span of `line` with a redacted form of the +/// value it contains, preserving the surrounding context (e.g. `api_key = `). +fn redact_span(line: &str, start: usize, end: usize) -> String { + let secret = &line[start..end]; + format!("{}{}{}", &line[..start], redact_value(secret), &line[end..]) +} + +/// Masks a secret value, keeping only the first/last 4 characters visible. +/// Short values (<=8 chars) are fully masked to avoid leaking most of a short +/// secret. +fn redact_value(value: &str) -> String { + let chars: Vec = value.chars().collect(); + let len = chars.len(); + + if len <= 8 { + return "*".repeat(len.max(4)); + } + + let first: String = chars[..4].iter().collect(); + let last: String = chars[len - 4..].iter().collect(); + format!("{}{}{}", first, "*".repeat(len - 8), last) +} + +/// A compiled `EntropyRule`, ready to scan file content without recompiling +/// its token regex on every call. +pub struct CompiledEntropyRule<'a> { + rule: &'a EntropyRule, + token_regex: Regex, +} + +/// Compiles all entropy rules in a `RuleSet` once per scan run (not once per +/// file), since regex compilation is comparatively expensive and files are +/// processed in parallel. +pub fn compile_entropy_rules(ruleset: &RuleSet) -> Vec { + ruleset + .entropy_rules + .iter() + .filter_map(|rule| { + Regex::new(&rule.token_pattern) + .ok() + .map(|token_regex| CompiledEntropyRule { rule, token_regex }) + }) + .collect() +} + +/// Scans file content for high-entropy tokens (e.g. long base64/hex blobs) +/// not covered by an explicit secret pattern. Always redacts matches. +pub fn scan_file_entropy( + file_path: &str, + content: &str, + compiled_rules: &[CompiledEntropyRule], + entropy_override: Option, +) -> Vec { + let mut issues = Vec::new(); + if compiled_rules.is_empty() { + return issues; + } + + let lines: Vec<&str> = content.lines().collect(); + + for entry in compiled_rules { + let rule = entry.rule; + + if let Some(file_pattern) = &rule.file_pattern { + if !wildmatch::WildMatch::new(file_pattern).matches(file_path) { + continue; + } + } + + let threshold = entropy_override.unwrap_or(rule.threshold); + + for (i, line) in lines.iter().enumerate() { + for m in entry.token_regex.find_iter(line) { + let token = m.as_str(); + if token.chars().count() < rule.min_length { + continue; + } + + if shannon_entropy(token) >= threshold { + let code = redact_span(line, m.start(), m.end()); + issues.push(Issue::new( + rule.id.clone(), + rule.description.clone(), + file_path.to_string(), + i + 1, + code, + rule.severity.clone(), + rule.confidence.clone(), + rule.remediation.clone(), + None, + )); + } + } + } + } + + issues +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::issues::Severity; + use crate::rules::Rule; + + fn redact_rule(id: &str, pattern: &str) -> Rule { + toml::from_str(&format!( + r#" + id = "{id}" + description = "test" + severity = "Critical" + pattern = '{pattern}' + redact = true + "# + )) + .unwrap() + } + + #[test] + fn redact_value_masks_middle_keeps_edges() { + assert_eq!(redact_value("AKIAABCDEFGH1234"), "AKIA********1234"); + } + + #[test] + fn redact_value_short_secret_fully_masked() { + let out = redact_value("abcd1234"); + assert_eq!(out, "*".repeat(8)); + } + + #[test] + fn scan_file_redacts_capture_group_only() { + let mut ruleset = RuleSet { + defaults: Default::default(), + rules: vec![redact_rule("SECRET001", r#"api_key\s*=\s*"([A-Za-z0-9]{12,})""#)], + taint_sources: vec![], + taint_sinks: vec![], + taint_sanitizers: vec![], + entropy_rules: vec![], + }; + ruleset.rules[0].severity = Severity::Critical; + + let content = "api_key = \"ABCDEFGHIJKL9999\"\n"; + let issues = scan_file("config.py", content, &ruleset); + + assert_eq!(issues.len(), 1); + assert!(issues[0].code.starts_with("api_key = \"ABCD")); + assert!(!issues[0].code.contains("ABCDEFGHIJKL9999")); + assert!(issues[0].code.contains("9999\"")); + } + + #[test] + fn scan_file_secret_rule_bypasses_string_literal_filter() { + let mut ruleset = RuleSet { + defaults: Default::default(), + rules: vec![redact_rule("SECRET002", r#""api_key":\s*"([A-Za-z0-9]{12,})""#)], + taint_sources: vec![], + taint_sinks: vec![], + taint_sanitizers: vec![], + entropy_rules: vec![], + }; + ruleset.rules[0].severity = Severity::Critical; + + // This line looks like a bare string literal to `is_in_comment_or_string` + // (starts and ends with a quote) but is a real JSON secret value. + let content = "\"api_key\": \"ABCDEFGHIJKL9999\"\n"; + let issues = scan_file("config.json", content, &ruleset); + + assert_eq!(issues.len(), 1); + } + + #[test] + fn looks_binary_detects_nul_byte() { + assert!(looks_binary(&[0x41, 0x42, 0x00, 0x43])); + assert!(!looks_binary(b"hello world")); + } + + #[test] + fn scan_file_entropy_flags_high_entropy_token_and_redacts() { + let mut ruleset = RuleSet { + defaults: Default::default(), + rules: vec![], + taint_sources: vec![], + taint_sinks: vec![], + taint_sanitizers: vec![], + entropy_rules: vec![], + }; + ruleset.entropy_rules.push(toml::from_str( + r#" + id = "ENTROPY001" + description = "High entropy token" + severity = "Medium" + threshold = 4.0 + min_length = 20 + "#, + ).unwrap()); + + let compiled = compile_entropy_rules(&ruleset); + let content = "token = 8f3kD9pQmZ2xN7vR1tYcL0aWs6HbUeJg\n"; + let issues = scan_file_entropy("config.py", content, &compiled, None); + + assert_eq!(issues.len(), 1); + assert!(!issues[0].code.contains("8f3kD9pQmZ2xN7vR1tYcL0aWs6HbUeJg")); + } + + #[test] + fn scan_file_entropy_ignores_low_entropy_text() { + let mut ruleset = RuleSet { + defaults: Default::default(), + rules: vec![], + taint_sources: vec![], + taint_sinks: vec![], + taint_sanitizers: vec![], + entropy_rules: vec![], + }; + ruleset.entropy_rules.push(toml::from_str( + r#" + id = "ENTROPY001" + description = "High entropy token" + severity = "Medium" + threshold = 4.5 + min_length = 20 + "#, + ).unwrap()); + + let compiled = compile_entropy_rules(&ruleset); + let content = "this_is_a_normal_variable_name_not_a_secret = True\n"; + let issues = scan_file_entropy("config.py", content, &compiled, None); + + assert_eq!(issues.len(), 0); + } +} diff --git a/src/pyspector/_rust_core/src/analysis/entropy.rs b/src/pyspector/_rust_core/src/analysis/entropy.rs new file mode 100644 index 0000000..e8f8e5c --- /dev/null +++ b/src/pyspector/_rust_core/src/analysis/entropy.rs @@ -0,0 +1,61 @@ +use std::collections::HashMap; + +/// Computes the Shannon entropy (in bits per character) of a string. +/// +/// High-entropy strings (random-looking base64/hex blobs) are a common +/// signature of secret material that isn't covered by an explicit pattern +/// (e.g. a bespoke API key format). Typical English text sits well under 4.0 +/// bits/char; base64-encoded random bytes sit close to 6.0. +pub fn shannon_entropy(s: &str) -> f64 { + let len = s.chars().count(); + if len == 0 { + return 0.0; + } + + let mut counts: HashMap = HashMap::new(); + for c in s.chars() { + *counts.entry(c).or_insert(0) += 1; + } + + let len_f = len as f64; + counts.values().fold(0.0, |acc, &count| { + let p = count as f64 / len_f; + acc - p * p.log2() + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn empty_string_has_zero_entropy() { + assert_eq!(shannon_entropy(""), 0.0); + } + + #[test] + fn single_repeated_char_has_zero_entropy() { + assert_eq!(shannon_entropy("aaaaaaaaaa"), 0.0); + } + + #[test] + fn low_entropy_english_word() { + // Highly repetitive/structured text should be well under 3 bits/char. + assert!(shannon_entropy("aaaabbbbcccc") < 2.5); + } + + #[test] + fn high_entropy_random_base64_like_token() { + // Random-looking base64 token should be high entropy (~5.5-6 bits/char). + let token = "8f3kD9pQmZ2xN7vR1tYcL0aWs6HbUeJg"; // pragma: allowlist secret + assert!(shannon_entropy(token) > 4.0); + } + + #[test] + fn four_char_alphabet_uniform_is_two_bits() { + // Perfectly uniform 4-symbol alphabet -> exactly 2 bits/char. + let s = "abcdabcdabcd"; + let e = shannon_entropy(s); + assert!((e - 2.0).abs() < 1e-9); + } +} diff --git a/src/pyspector/_rust_core/src/analysis/mod.rs b/src/pyspector/_rust_core/src/analysis/mod.rs index af75a41..7d49a30 100644 --- a/src/pyspector/_rust_core/src/analysis/mod.rs +++ b/src/pyspector/_rust_core/src/analysis/mod.rs @@ -20,14 +20,16 @@ fn severity_rank(s: &Severity) -> u8 { } mod ast_analysis; -mod config_analysis; +pub mod config_analysis; mod taint_analysis; +pub mod entropy; pub struct AnalysisContext<'a> { pub root_path: String, pub exclusions: Vec, pub ruleset: RuleSet, pub py_files: &'a [PythonFile], + pub entropy_threshold: Option, } pub fn run_analysis(mut context: AnalysisContext) -> Vec { @@ -43,10 +45,10 @@ pub fn run_analysis(mut context: AnalysisContext) -> Vec { } } println!("[*] Starting analysis with {} rules", context.ruleset.rules.len()); - + let root_path = Path::new(&context.root_path); let mut files_to_scan: Vec = Vec::new(); - + // Add common test fixture patterns to exclusions let mut enhanced_exclusions = context.exclusions.clone(); enhanced_exclusions.extend(vec![ @@ -55,11 +57,6 @@ pub fn run_analysis(mut context: AnalysisContext) -> Vec { "*_test.py".to_string(), "*/test_*.py".to_string(), ]); - // Precompile each exclusion pattern once — is_excluded() is called once per - // file in the walk below and once per file again in the AST pass, so - // re-parsing every glob pattern with WildMatch::new() on every call (and - // twice per call, at that) adds up on large codebases. - let compiled_exclusions = compile_exclusions(&enhanced_exclusions); for entry in WalkDir::new(root_path).into_iter().filter_map(|e| e.ok()) { let path = entry.path(); @@ -70,20 +67,34 @@ pub fn run_analysis(mut context: AnalysisContext) -> Vec { } } } - + println!("[+] Found {} files to scan ({} non-Python)", files_to_scan.len(), files_to_scan.iter().filter(|f| !f.ends_with(".py")).count()); - // Scan all files with regex patterns + // Entropy rules' token regexes are compiled once per run, not once per file. + let compiled_entropy_rules = config_analysis::compile_entropy_rules(&context.ruleset); + + // Scan all files with regex + entropy patterns let t_config = std::time::Instant::now(); let mut issues: Vec = files_to_scan .par_iter() .flat_map(|file_path| { - if let Ok(content) = fs::read_to_string(file_path) { - config_analysis::scan_file(file_path, &content, &context.ruleset) - } else { - Vec::new() + let Ok(bytes) = fs::read(file_path) else { + return Vec::new(); + }; + if config_analysis::looks_binary(&bytes) { + return Vec::new(); } + let content = String::from_utf8_lossy(&bytes); + + let mut findings = config_analysis::scan_file(file_path, &content, &context.ruleset); + findings.extend(config_analysis::scan_file_entropy( + file_path, + &content, + &compiled_entropy_rules, + context.entropy_threshold, + )); + findings }) .collect(); println!("[*] Pattern/config scan: {:.2}s → {} issues", t_config.elapsed().as_secs_f64(), issues.len()); @@ -114,7 +125,7 @@ pub fn run_analysis(mut context: AnalysisContext) -> Vec { let taint_issues = taint_analysis::analyze_program_for_taint(&call_graph, &context.ruleset); println!("[+] Found {} issues from taint analysis", taint_issues.len()); issues.extend(taint_issues); - + let mut seen = HashSet::new(); issues.retain(|issue| seen.insert(issue.get_fingerprint())); @@ -159,18 +170,15 @@ pub fn run_analysis(mut context: AnalysisContext) -> Vec { issues } -/// Either a precompiled glob pattern or a plain substring, decided once at -/// compile time (see `compile_exclusions`) instead of re-inspecting the raw -/// string and reparsing the glob on every file checked. -enum ExclusionPattern { - Glob(wildmatch::WildMatch), - Substring(String), -} +fn is_excluded(path: &Path, exclusions: &[String]) -> bool { + let path_str = path.to_str().unwrap_or_default(); + let path_filename = path.file_name().and_then(|s| s.to_str()).unwrap_or_default(); -fn compile_exclusions(exclusions: &[String]) -> Vec { - exclusions.iter().map(|ex| { + exclusions.iter().any(|ex| { + // Handle glob patterns if ex.contains('*') { - ExclusionPattern::Glob(wildmatch::WildMatch::new(ex)) + wildmatch::WildMatch::new(ex).matches(path_str) || + wildmatch::WildMatch::new(ex).matches(path_filename) } else { ExclusionPattern::Substring(ex.clone()) } diff --git a/src/pyspector/_rust_core/src/lib.rs b/src/pyspector/_rust_core/src/lib.rs index 4a18b0f..b49b664 100644 --- a/src/pyspector/_rust_core/src/lib.rs +++ b/src/pyspector/_rust_core/src/lib.rs @@ -1,17 +1,18 @@ use pyo3::prelude::*; use pyo3::types::{PyDict, PyList}; use rayon::prelude::*; +use std::collections::HashSet; mod ast_parser; mod graph; -mod issues; -mod rules; -mod analysis; +pub mod issues; +pub mod rules; +pub mod analysis; mod supply_chain; use issues::{Issue, Severity}; use rules::RuleSet; -use analysis::{run_analysis, AnalysisContext}; +use analysis::{run_analysis, AnalysisContext, config_analysis}; use ast_parser::PythonFile; #[pyfunction] @@ -23,8 +24,9 @@ fn run_scan_py<'py>( config: &Bound<'py, PyDict>, python_files_data: &Bound<'py, PyList>, ) -> PyResult> { - + let exclusions: Vec = config.get_item("exclude")?.map_or(Ok(Vec::new()), |v| v.extract())?; + let entropy_threshold: Option = config.get_item("entropy_threshold")?.map_or(Ok(None), |v| v.extract())?; let mut ruleset: RuleSet = toml::from_str(&rules_toml_str).map_err(|e| { pyo3::exceptions::PyValueError::new_err(format!("Failed to parse rules: {}", e)) @@ -47,6 +49,14 @@ fn run_scan_py<'py>( raw_files.push((file_path, content, ast_json)); } + let context = AnalysisContext { + root_path: path, + exclusions, + ruleset, + py_files: &py_files, + entropy_threshold, + }; + // PyO3 renamed `allow_threads` to `detach` let issues = py.detach(|| { // Parse every file's AST JSON in parallel across all cores instead of @@ -71,7 +81,71 @@ fn run_scan_py<'py>( for issue in issues { py_issues.append(Py::new(py, issue)?)?; } - + + Ok(py_issues) +} + +#[pyfunction] +#[pyo3(name = "scan_blobs")] +fn scan_blobs_py<'py>( + py: Python<'py>, + blobs: &Bound<'py, PyList>, + rules_toml_str: String, + config: &Bound<'py, PyDict>, +) -> PyResult> { + let entropy_threshold: Option = + config.get_item("entropy_threshold")?.map_or(Ok(None), |v| v.extract())?; + + let ruleset: RuleSet = toml::from_str(&rules_toml_str).map_err(|e| { + pyo3::exceptions::PyValueError::new_err(format!("Failed to parse rules: {}", e)) + })?; + + struct Blob { + label: String, + content: String, + } + + let mut items: Vec = Vec::new(); + for item in blobs.iter() { + let dict: Bound<'py, PyDict> = item.extract()?; + let path: String = dict.get_item("path")?.unwrap().extract()?; + let content: String = dict.get_item("content")?.unwrap().extract()?; + let commit: String = dict.get_item("commit")?.unwrap().extract()?; + items.push(Blob { + label: format!("{}:{}", commit, path), + content, + }); + } + + let compiled_entropy_rules = config_analysis::compile_entropy_rules(&ruleset); + + // PyO3 renamed `allow_threads` to `detach` + let issues: Vec = py.detach(|| { + let mut all: Vec = items + .par_iter() + .flat_map(|blob| { + let mut findings = + config_analysis::scan_file(&blob.label, &blob.content, &ruleset); + findings.extend(config_analysis::scan_file_entropy( + &blob.label, + &blob.content, + &compiled_entropy_rules, + entropy_threshold, + )); + findings + }) + .collect(); + + let mut seen = HashSet::new(); + all.retain(|issue| seen.insert(issue.get_fingerprint())); + all + }); + + let py_issues = PyList::empty(py); + for issue in issues { + py_issues.append(Py::new(py, issue)?)?; + } + Ok(py_issues) } @@ -103,12 +177,13 @@ fn scan_supply_chain_py<'py>( } #[pymodule] -fn _rust_core(m: &Bound<'_, PyModule>) -> PyResult<()> { +fn _rust_core(m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; m.add_class::()?; m.add_function(wrap_pyfunction!(run_scan_py, m)?)?; + m.add_function(wrap_pyfunction!(scan_blobs_py, m)?)?; m.add_function(wrap_pyfunction!(scan_supply_chain_py, m)?)?; Ok(()) -} \ No newline at end of file +} diff --git a/src/pyspector/_rust_core/src/rules.rs b/src/pyspector/_rust_core/src/rules.rs index 1463eba..67461fc 100644 --- a/src/pyspector/_rust_core/src/rules.rs +++ b/src/pyspector/_rust_core/src/rules.rs @@ -64,6 +64,13 @@ pub struct Rule { /// output for downstream tooling. #[serde(default)] pub cwe: Option, + /// When true, this rule matches secret material: the matched value (or its + /// first capture group, if any) is redacted before the match is stored on + /// an `Issue`, and the Python-literal comment/string heuristic in + /// `config_analysis::is_in_comment_or_string` is bypassed, since secrets + /// routinely live inside string literals (.env/.json/.yaml values). + #[serde(default)] + pub redact: bool, } impl Rule { @@ -108,6 +115,28 @@ impl Rule { fn default_confidence() -> String { "Medium".to_string() } +fn default_token_pattern() -> String { r"[A-Za-z0-9+/_=-]{20,}".to_string() } + +fn default_min_length() -> usize { 20 } + +#[derive(Debug, Deserialize, Clone)] +pub struct EntropyRule { + pub id: String, + pub description: String, + pub severity: Severity, + #[serde(default = "default_confidence")] + pub confidence: String, + #[serde(default)] + pub remediation: String, + #[serde(default)] + pub file_pattern: Option, + #[serde(default = "default_token_pattern")] + pub token_pattern: String, + #[serde(default = "default_min_length")] + pub min_length: usize, + pub threshold: f64, +} + #[derive(Debug, Deserialize)] pub struct TaintSourceRule { pub id: String, @@ -203,4 +232,4 @@ impl RuleSet { .unwrap_or_default(); } } -} \ No newline at end of file +} diff --git a/src/pyspector/_rust_core/tests/secrets_patterns.rs b/src/pyspector/_rust_core/tests/secrets_patterns.rs new file mode 100644 index 0000000..bb488d1 --- /dev/null +++ b/src/pyspector/_rust_core/tests/secrets_patterns.rs @@ -0,0 +1,178 @@ +//! Table-driven tests for the built-in secret-detection rules +//! (`src/pyspector/rules/built-in-rules-secrets.toml`): one synthetic positive +//! fixture and one negative (must-not-match) fixture per rule id, plus a +//! check that redacted rules never store the raw secret verbatim. + +use _rust_core::analysis::config_analysis::{compile_entropy_rules, scan_file, scan_file_entropy}; +use _rust_core::rules::RuleSet; + +const SECRETS_TOML: &str = include_str!("../../rules/built-in-rules-secrets.toml"); + +fn ruleset() -> RuleSet { + toml::from_str(SECRETS_TOML).expect("built-in-rules-secrets.toml should parse") +} + +struct Case { + rule_id: &'static str, + file_path: &'static str, + positive: &'static str, + negative: &'static str, +} + +const CASES: &[Case] = &[ + Case { + rule_id: "SEC-AWS-001", + file_path: "config.py", + // Split across concat! so the literal AWS-shaped key never appears + // contiguous in source (avoids tripping GitHub push protection). + positive: concat!("aws_key = \"AKIA", "ABCDEFGHIJKLMNOP\""), + negative: "aws_key = \"AKIA123\"", + }, + Case { + rule_id: "SEC-AWS-002", + file_path: "config.py", + positive: concat!( + "aws_secret_access_key = \"abcdefghijklmnopqrstuvwxyz", + "ABCDEFGHIJ1234\"" + ), + negative: "aws_secret_access_key = \"short\"", + }, + Case { + rule_id: "SEC-GCP-001", + file_path: "config.json", + positive: "\"type\": \"service_account\"", + negative: "\"type\": \"authorized_user\"", + }, + Case { + rule_id: "SEC-GCP-002", + file_path: "config.json", + positive: "\"private_key\": \"-----BEGIN PRIVATE KEY-----FAKEKEYDATA-----END PRIVATE KEY-----\"", // pragma: allowlist secret + negative: "\"other_key\": \"not-a-key\"", + }, + Case { + rule_id: "SEC-AZURE-001", + file_path: "config.py", + positive: "conn = \"DefaultEndpointsProtocol=https;AccountName=myaccount;AccountKey=abcdEFGH1234567890abcdEFGH1234567890AB==\"", + negative: "conn = \"DefaultEndpointsProtocol=https;AccountName=myaccount\"", + }, + Case { + rule_id: "SEC-GITHUB-001", + file_path: "config.py", + positive: "token = \"ghp_1234567890ABCDEFGHIJKLMNOPQRSTUVWXYZ\"", // pragma: allowlist secret + negative: "token = \"ghp_short\"", + }, + Case { + rule_id: "SEC-SLACK-001", + file_path: "config.py", + positive: concat!( + "token = \"xoxb-123456789012-1234567890123", + "-abcdefghijklmnopqrstuvwx\"" // pragma: allowlist secret + ), + negative: "token = \"xoxb-short\"", + }, + Case { + rule_id: "SEC-STRIPE-001", + file_path: "config.py", + positive: concat!("key = \"sk_liv", "e_abcdefghijklmnopqrstuvwx1234\""), // pragma: allowlist secret + negative: "key = \"sk_test_abcdefghijklmnopqrstuvwx1234\"", // pragma: allowlist secret + }, + Case { + rule_id: "SEC-PRIVATEKEY-001", + file_path: "id_rsa", + positive: "-----BEGIN RSA PRIVATE KEY-----", // pragma: allowlist secret + negative: "-----BEGIN CERTIFICATE-----", + }, + Case { + rule_id: "SEC-JWT-001", + file_path: "config.py", + positive: "token = \"eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0.dozjgNryP4J3jVmNHl0w5N_XgL0n3I9PYb4LddY2Tm4\"", // pragma: allowlist secret + negative: "token = \"eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0\"", // pragma: allowlist secret + }, + Case { + rule_id: "SEC-GENERIC-001", + file_path: "config.py", + positive: "password = \"Sup3rSecretValue123\"", + negative: "password = \"abc\"", + }, +]; + +#[test] +fn secret_patterns_match_positive_and_not_negative_fixtures() { + let rs = ruleset(); + let known_ids: Vec<&str> = rs.rules.iter().map(|r| r.id.as_str()).collect(); + + for case in CASES { + assert!( + known_ids.contains(&case.rule_id), + "rule {} is not present in built-in-rules-secrets.toml (test/rule id drift)", + case.rule_id + ); + + let pos_issues = scan_file(case.file_path, case.positive, &rs); + assert!( + pos_issues.iter().any(|i| i.rule_id == case.rule_id), + "expected rule {} to match positive fixture: {:?}", + case.rule_id, + case.positive + ); + + let neg_issues = scan_file(case.file_path, case.negative, &rs); + assert!( + !neg_issues.iter().any(|i| i.rule_id == case.rule_id), + "expected rule {} NOT to match negative fixture: {:?}", + case.rule_id, + case.negative + ); + } +} + +#[test] +fn redacted_rules_never_store_the_raw_secret_line() { + let rs = ruleset(); + + for case in CASES { + let rule = rs + .rules + .iter() + .find(|r| r.id == case.rule_id) + .expect("rule must exist (checked in previous test)"); + + if !rule.redact { + continue; + } + + let issues = scan_file(case.file_path, case.positive, &rs); + let matches: Vec<_> = issues.iter().filter(|i| i.rule_id == case.rule_id).collect(); + assert!(!matches.is_empty(), "rule {} produced no issues", case.rule_id); + + for issue in matches { + assert_ne!( + issue.code, case.positive, + "rule {} is marked redact=true but stored the raw, unredacted line", + case.rule_id + ); + } + } +} + +#[test] +fn entropy_rule_flags_random_token_but_not_repetitive_text() { + let rs = ruleset(); + let compiled = compile_entropy_rules(&rs); + assert!(!compiled.is_empty(), "expected at least one entropy_rule in built-in-rules-secrets.toml"); + + let content_secret = "value = \"Zx9Qk3mP7vT2wL8dR4nC6yB1sA5jF0hE\"\n"; // pragma: allowlist secret + let issues = scan_file_entropy("config.py", content_secret, &compiled, None); + assert!(!issues.is_empty(), "expected entropy rule to flag a high-entropy token"); + assert!( + !issues[0].code.contains("Zx9Qk3mP7vT2wL8dR4nC6yB1sA5jF0hE"), + "entropy findings must be redacted" + ); + + let content_plain = "value = \"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\"\n"; + let issues2 = scan_file_entropy("config.py", content_plain, &compiled, None); + assert!( + issues2.is_empty(), + "expected entropy rule not to flag low-entropy repeated text" + ); +} diff --git a/src/pyspector/cli.py b/src/pyspector/cli.py index 62e442b..c453ace 100644 --- a/src/pyspector/cli.py +++ b/src/pyspector/cli.py @@ -20,13 +20,13 @@ from .reporting import Reporter from .triage import run_triage_tui from .stats import StatsCollector -from .messages import handle_msg_flag +from .git_history import iter_all_blobs, iter_staged_files import requests from urllib.parse import urlparse # Import the Rust core from its new location try: - from pyspector._rust_core import run_scan + from pyspector._rust_core import run_scan, scan_blobs except ImportError: click.echo(click.style("Error: PySpector's core engine module not found.", fg="red")) exit(1) @@ -945,6 +945,271 @@ def triage_command(report_file: Path): click.echo(click.style(f"Error reading report file: {e}", fg="red")) +# --- Secret Detection Commands --- + +def _load_ignored_fingerprints(baseline_path: Path) -> set: + """Loads the set of ignored fingerprints from a .pyspector_baseline.json file.""" + if not baseline_path.exists(): + return set() + try: + with baseline_path.open('r') as f: + data = json.load(f) + return set(data.get("ignored_fingerprints", [])) + except (json.JSONDecodeError, IOError): + return set() + + +def _run_secrets_scan( + scan_path: Path, + config_path: Optional[Path], + history: bool, + staged_only: bool, + entropy_threshold: Optional[float], +) -> list: + """ + Runs the secret-detection ruleset over the working tree and/or git + history/staged content. Returns a fingerprint-deduplicated list of Issue + objects (before baseline/severity filtering). + """ + config = load_config(config_path) + if entropy_threshold is not None: + config['entropy_threshold'] = entropy_threshold + + rules_toml_str = get_default_rules(secrets_scan=True) + + all_issues = [] + + if staged_only: + blobs = [ + {"path": path, "content": content, "commit": "staged"} + for path, content in iter_staged_files(scan_path) + ] + if blobs: + all_issues.extend(scan_blobs(blobs, rules_toml_str, config)) + else: + # Working tree: no Python ASTs are needed (secrets are regex/entropy only), + # which keeps this scan fast even on large repos. + all_issues.extend(run_scan(str(scan_path.resolve()), rules_toml_str, config, [])) + + if history: + click.echo("[*] Scanning full git history for secrets (this may take a while on large repos)...") + blobs = [ + {"path": path, "content": content, "commit": "history"} + for path, content in iter_all_blobs(scan_path) + ] + if blobs: + all_issues.extend(scan_blobs(blobs, rules_toml_str, config)) + + seen = set() + deduped = [] + for issue in all_issues: + fingerprint = issue.get_fingerprint() + if fingerprint not in seen: + seen.add(fingerprint) + deduped.append(issue) + + return deduped + + +@click.group(help="Detect hardcoded secrets in the working tree, staged changes, or full git history.") +def secrets(): + """Secret detection commands.""" + pass + + +@secrets.command(name="scan", help="Scan for hardcoded secrets.") +@click.argument('path', type=click.Path(exists=True, file_okay=False, dir_okay=True, path_type=Path), required=False, default=Path('.')) +@click.option('-c', '--config', 'config_path', type=click.Path(exists=True, path_type=Path), help="Path to a pyspector config TOML file.") +@click.option('-o', '--output', 'output_file', type=click.Path(path_type=Path), help="Path to write the report to.") +@click.option('-f', '--format', 'report_format', type=click.Choice(['console', 'json', 'sarif', 'html']), default='console', help="Format of the report.") +@click.option('-s', '--severity', 'severity_level', type=click.Choice(['LOW', 'MEDIUM', 'HIGH', 'CRITICAL']), default='LOW', help="Minimum severity level to report.") +@click.option('--history', is_flag=True, default=False, help="Also scan the full git history (all commits), not just the working tree.") +@click.option('--staged-only', is_flag=True, default=False, help="Scan only staged (git diff --cached) content. Intended for pre-commit hooks.") +@click.option('--entropy-threshold', type=float, default=None, help="Override the Shannon-entropy threshold (bits/char) for the high-entropy rule.") +@click.option('--fail-on-findings/--no-fail-on-findings', 'fail_on_findings', default=True, help="Exit non-zero if unbaselined findings remain (default: enabled, for CI use).") +def secrets_scan_command( + path: Path, + config_path: Optional[Path], + output_file: Optional[Path], + report_format: str, + severity_level: str, + history: bool, + staged_only: bool, + entropy_threshold: Optional[float], + fail_on_findings: bool, +): + """Scan for hardcoded secrets and exit non-zero if any unbaselined findings remain.""" + if staged_only and history: + raise click.UsageError("--staged-only and --history cannot be used together.") + + start_time = time.time() + click.echo(f"[*] Starting PySpector secret scan on '{path}'...") + + issues = _run_secrets_scan(path, config_path, history, staged_only, entropy_threshold) + + baseline_path = path / ".pyspector_baseline.json" + ignored_fingerprints = _load_ignored_fingerprints(baseline_path) + if ignored_fingerprints: + click.echo(f"[*] Loaded baseline from '{baseline_path}', ignoring {len(ignored_fingerprints)} known finding(s).") + + severity_map = {'LOW': 0, 'MEDIUM': 1, 'HIGH': 2, 'CRITICAL': 3} + min_severity_val = severity_map[severity_level.upper()] + + final_issues = [ + issue for issue in issues + if (severity_map[str(issue.severity).split('.')[-1].upper()] >= min_severity_val + and issue.get_fingerprint() not in ignored_fingerprints) + ] + + reporter = Reporter(final_issues, report_format) + output = reporter.generate() + + if output_file: + try: + output_file.write_text(output, encoding='utf-8') + click.echo(f"\n[+] Report saved to '{output_file}'") + except IOError as e: + click.echo(click.style(f"Error writing to output file: {e}", fg="red")) + else: + click.echo(output) + + end_time = time.time() + click.echo(f"\n[*] Secret scan finished in {end_time - start_time:.2f} seconds. Found {len(final_issues)} unbaselined issue(s).") + if len(issues) > len(final_issues): + click.echo(f"[*] Ignored {len(issues) - len(final_issues)} finding(s) based on severity level or baseline.") + + sys.stdout.flush() + sys.stderr.flush() + + if fail_on_findings and final_issues: + sys.exit(1) + + +@secrets.command(name="baseline", help="Snapshot current secret findings into .pyspector_baseline.json for incremental adoption on legacy repos.") +@click.argument('path', type=click.Path(exists=True, file_okay=False, dir_okay=True, path_type=Path), required=False, default=Path('.')) +@click.option('-c', '--config', 'config_path', type=click.Path(exists=True, path_type=Path), help="Path to a pyspector config TOML file.") +@click.option('--history', is_flag=True, default=False, help="Also include findings from the full git history in the baseline.") +@click.option('--entropy-threshold', type=float, default=None, help="Override the Shannon-entropy threshold (bits/char) for the high-entropy rule.") +def secrets_baseline_command(path: Path, config_path: Optional[Path], history: bool, entropy_threshold: Optional[float]): + """Non-interactively baseline all current secret findings (does not fail/exit non-zero).""" + click.echo(f"[*] Scanning '{path}' to build a secrets baseline...") + + issues = _run_secrets_scan(path, config_path, history, staged_only=False, entropy_threshold=entropy_threshold) + + baseline_path = path / ".pyspector_baseline.json" + existing_fingerprints = _load_ignored_fingerprints(baseline_path) + fingerprints = existing_fingerprints | {issue.get_fingerprint() for issue in issues} + + baseline_path.write_text( + json.dumps({"ignored_fingerprints": sorted(fingerprints)}, indent=2), + encoding='utf-8', + ) + + click.echo(click.style( + f"[+] Baseline saved to '{baseline_path}' with {len(fingerprints)} ignored finding(s).", + fg="green", + )) + click.echo("[*] Run 'pyspector triage ' to interactively review or adjust the baseline later.") + + +_SECRETS_PRE_COMMIT_HOOK = """#!/bin/bash + +# PySpector secret-detection pre-commit hook + +echo "[PySpector] Scanning staged changes for secrets..." + +pyspector secrets scan --staged-only --severity LOW + +SCAN_RESULT=$? + +if [ $SCAN_RESULT -ne 0 ]; then + echo "" + echo "[PySpector] Commit aborted: secret(s) detected in staged changes." + echo "[PySpector] Review the findings above, remove the secret(s), or add them to .pyspector_baseline.json if they are false positives (see 'pyspector secrets baseline')." + echo "[PySpector] To bypass in an emergency: git commit --no-verify (not recommended)." + exit 1 +fi + +echo "[PySpector] No secrets found. Proceeding with commit." +exit 0 +""" + + +@secrets.command(name="install-hook", help="Install a Git pre-commit hook that runs 'pyspector secrets scan --staged-only' before every commit.") +@click.option('--force', is_flag=True, help="Overwrite an existing pre-commit hook.") +def secrets_install_hook_command(force: bool): + """Installs the secret-detection pre-commit hook into the current repository.""" + try: + repo_root = Path(subprocess.run( + ['git', 'rev-parse', '--show-toplevel'], + check=True, capture_output=True, text=True, + ).stdout.strip()) + except (subprocess.CalledProcessError, FileNotFoundError): + click.echo(click.style("Error: Not inside a Git repository.", fg="red")) + sys.exit(1) + + hook_path = repo_root / ".git" / "hooks" / "pre-commit" + + if hook_path.exists() and not force: + click.echo(click.style(f"Error: '{hook_path}' already exists. Use --force to overwrite.", fg="red")) + sys.exit(1) + + hook_path.parent.mkdir(parents=True, exist_ok=True) + hook_path.write_text(_SECRETS_PRE_COMMIT_HOOK, encoding='utf-8', newline='\n') + hook_path.chmod(hook_path.stat().st_mode | 0o111) + + click.echo(click.style(f"[+] Installed secret-detection pre-commit hook at '{hook_path}'.", fg="green")) + + +# --- Plugin Management Commands --- + +@click.group(help="Manage PySpector plugins") +def plugin(): + """Plugin management commands""" + pass + + +@plugin.command(name="list", help="List all available plugins") +def list_plugins_command(): + """List available plugins""" + plugin_manager = get_plugin_manager() + available = plugin_manager.list_available_plugins() + registered = plugin_manager.registry.list_plugins() + + click.echo("\n" + "="*60) + click.echo("PySpector Plugins") + click.echo("="*60) + + if not available: + click.echo("\nNo plugins found in plugin directory") + click.echo(f"Plugin directory: {plugin_manager.plugin_dir}") + else: + click.echo(f"\nFound {len(available)} plugin(s):\n") + + for plugin_name in available: + info = next((p for p in registered if p["name"] == plugin_name), None) + + if info: + is_trusted = bool(info.get("trusted")) + status_text = "trusted" if is_trusted else "untrusted" + status_color = "green" if is_trusted else "yellow" + status = click.style(status_text, fg=status_color) + click.echo(f" {plugin_name}") + click.echo(f" Status: {status}") + click.echo(f" Version: {info.get('version', 'unknown')}") + click.echo(f" Author: {info.get('author', 'unknown')}") + click.echo(f" Category: {info.get('category', 'general')}") + else: + click.echo(f" {plugin_name}") + click.echo( + f" Status: {click.style('not registered', fg='red')}" + ) + + click.echo() + + click.echo(f"Plugin directory: {plugin_manager.plugin_dir}") + click.echo("="*60 + "\n") + @click.command( help=( @@ -1154,4 +1419,5 @@ def on_moved(self, event) -> None: self._dispatch(event) # Add commands to the CLI group cli.add_command(run_scan_command, name="scan") cli.add_command(triage_command, name="triage") -cli.add_command(watch_command, name="watch") +cli.add_command(plugin) +cli.add_command(secrets) diff --git a/src/pyspector/config.py b/src/pyspector/config.py index b850f82..3381109 100644 --- a/src/pyspector/config.py +++ b/src/pyspector/config.py @@ -39,28 +39,57 @@ "severity": "LOW", } +# Default Shannon-entropy threshold (bits/char) for the generic high-entropy +# secret rule. Can be overridden via `[tool.pyspector.secrets] entropy_threshold` +# in the config file, or the `--entropy-threshold` CLI flag. +DEFAULT_ENTROPY_THRESHOLD = 4.5 + def load_config(config_path: Path) -> dict: - """Loads configuration from a TOML file or returns defaults.""" + """Loads configuration from a TOML file or returns defaults. + + A `[tool.pyspector.secrets]` sub-table (if present) is merged into the + returned config: its `entropy_threshold` overrides the top-level + `entropy_threshold` key (read directly by the Rust scan engine), and its + `exclude` list is appended to the main exclude list. + """ + config = DEFAULT_CONFIG.copy() + config["entropy_threshold"] = DEFAULT_ENTROPY_THRESHOLD + if config_path and config_path.exists(): try: with config_path.open('r') as f: user_config = toml.load(f).get('tool', {}).get('pyspector', {}) - config = deepcopy(DEFAULT_CONFIG) + secrets_config = user_config.pop('secrets', {}) if isinstance(user_config, dict) else {} + config.update(user_config) + + if 'entropy_threshold' in secrets_config: + config['entropy_threshold'] = secrets_config['entropy_threshold'] + if 'exclude' in secrets_config: + config['exclude'] = list(config.get('exclude', [])) + list(secrets_config['exclude']) + return config except Exception as e: click.echo(click.style(f"Warning: Could not parse config file '{config_path}'. Using defaults. Error: {e}", fg="yellow")) - return deepcopy(DEFAULT_CONFIG) + return config -def get_default_rules(ai_scan: bool = False) -> str: +def get_default_rules(ai_scan: bool = False, secrets_scan: bool = False) -> str: """Loads the built-in TOML rules file from package resources. Substitutes the `__SHARED_PLACEHOLDERS__` sentinel inside any rule's exclude_pattern with the value of `[defaults].exclude_pattern_placeholder`, so the placeholder/dummy-secret regex lives in one place rather than being copy-pasted across every format-specific rule. + + When `secrets_scan` is True, only the secret-detection ruleset is + returned (not the main/AI rulesets): secret scans skip Python AST parsing + entirely and only need regex + entropy rules, so keeping the ruleset + focused avoids unrelated noise and keeps the scan fast. """ try: + if secrets_scan: + return pkg_resources.files('pyspector.rules').joinpath('built-in-rules-secrets.toml').read_text(encoding='utf-8') + base_rules = pkg_resources.files('pyspector.rules').joinpath('built-in-rules.toml').read_text(encoding='utf-8') if ai_scan: click.echo("[*] AI scanning enabled. Loading additional AI/LLM rules.") @@ -75,4 +104,4 @@ def get_default_rules(ai_scan: bool = False) -> str: text = text.replace(_PLACEHOLDER_SENTINEL, m.group(1)) return text except Exception as e: - raise FileNotFoundError(f"Could not load built-in-rules.toml from package data! Error: {e}") + raise FileNotFoundError(f"Could not load built-in rules from package data! Error: {e}") diff --git a/src/pyspector/git_history.py b/src/pyspector/git_history.py new file mode 100644 index 0000000..b0aa750 --- /dev/null +++ b/src/pyspector/git_history.py @@ -0,0 +1,143 @@ +""" +Git history and staged-content helpers for PySpector secret detection. + +Scanning the working tree alone misses secrets that were committed and later +"removed" (they still live in the object database). `iter_all_blobs` walks +every blob reachable from any ref, scanning each unique blob's content +exactly once regardless of how many commits/paths reference it, using a +single `git rev-list` + two `git cat-file` batch calls rather than one +subprocess per blob or per commit. +""" + +import subprocess +from pathlib import Path +from typing import Dict, Iterator, List, Tuple + +MAX_BLOB_SIZE = 5 * 1024 * 1024 # 5MB - skip anything larger (unlikely to be a secret file) + + +def _run_git(args: List[str], cwd: Path, input_bytes: bytes = None) -> subprocess.CompletedProcess: + return subprocess.run( + ["git", "-C", str(cwd), *args], + input=input_bytes, + capture_output=True, + check=True, + ) + + +def _looks_binary(data: bytes) -> bool: + return b"\x00" in data[:8192] + + +def iter_all_blobs(repo_path: Path) -> Iterator[Tuple[str, str]]: + """ + Yields (path, text_content) for every blob reachable from any ref in the + repository's history, each scanned exactly once regardless of how many + commits reference identical content. Binary and oversized blobs are + skipped. Yields nothing if `repo_path` is not a git repository or `git` + is unavailable. + """ + try: + rev_list = _run_git(["rev-list", "--objects", "--all"], repo_path) + except (subprocess.CalledProcessError, FileNotFoundError): + return + + blob_paths: Dict[str, str] = {} + for line in rev_list.stdout.decode("utf-8", "replace").splitlines(): + if not line: + continue + sha, _, path = line.partition(" ") + if path and sha not in blob_paths: + blob_paths[sha] = path + + if not blob_paths: + return + + shas = list(blob_paths.keys()) + + try: + check_result = _run_git( + ["cat-file", "--batch-check=%(objectname) %(objecttype) %(objectsize)"], + repo_path, + input_bytes=("\n".join(shas) + "\n").encode("utf-8"), + ) + except subprocess.CalledProcessError: + return + + blob_shas: List[str] = [] + for line in check_result.stdout.decode("utf-8", "replace").splitlines(): + parts = line.split(" ") + if len(parts) != 3: + continue + sha, obj_type, size_str = parts + if obj_type == "blob" and size_str.isdigit() and int(size_str) <= MAX_BLOB_SIZE: + blob_shas.append(sha) + + if not blob_shas: + return + + try: + batch_result = _run_git( + ["cat-file", "--batch"], + repo_path, + input_bytes=("\n".join(blob_shas) + "\n").encode("utf-8"), + ) + except subprocess.CalledProcessError: + return + + data = batch_result.stdout + pos = 0 + while pos < len(data): + newline_idx = data.find(b"\n", pos) + if newline_idx == -1: + break + + header = data[pos:newline_idx].decode("utf-8", "replace") + pos = newline_idx + 1 + + parts = header.split(" ") + if len(parts) != 3 or not parts[2].isdigit(): + continue + + sha, obj_type, size_str = parts + size = int(size_str) + content_bytes = data[pos : pos + size] + pos += size + 1 # skip the record's trailing newline + + if obj_type != "blob" or _looks_binary(content_bytes): + continue + + path = blob_paths.get(sha) + if not path: + continue + + yield path, content_bytes.decode("utf-8", "replace") + + +def iter_staged_files(repo_path: Path) -> Iterator[Tuple[str, str]]: + """ + Yields (path, text_content) for the staged (index) version of each + added/copied/modified file, used by the pre-commit hook to scan exactly + what is about to be committed rather than the working-tree copy. + """ + try: + result = _run_git( + ["diff", "--cached", "--name-only", "--diff-filter=ACM"], + repo_path, + ) + except (subprocess.CalledProcessError, FileNotFoundError): + return + + paths = [p for p in result.stdout.decode("utf-8", "replace").splitlines() if p] + + for path in paths: + try: + show_result = _run_git(["show", f":{path}"], repo_path) + except subprocess.CalledProcessError: + continue + + content_bytes = show_result.stdout + if _looks_binary(content_bytes): + continue + + yield path, content_bytes.decode("utf-8", "replace") diff --git a/src/pyspector/rules/built-in-rules-secrets.toml b/src/pyspector/rules/built-in-rules-secrets.toml new file mode 100644 index 0000000..f25a351 --- /dev/null +++ b/src/pyspector/rules/built-in-rules-secrets.toml @@ -0,0 +1,146 @@ +# PySpector Secret Detection Rules +# +# All `[[rule]]` entries here set `redact = true`: the matched secret value +# (its first capture group, if the pattern has one) is masked to first/last 4 +# characters before it is ever stored on an `Issue`, so raw secret material +# never reaches a report, log, or terminal. `redact = true` also makes the +# rule bypass the Python comment/string-literal heuristic in +# `config_analysis::is_in_comment_or_string`, since secrets routinely live +# inside string literals (.env/.json/.yaml values). + +# ------------------------------------------- +# SECTION: Cloud provider credentials +# ------------------------------------------- + +[[rule]] +id = "SEC-AWS-001" +description = "Hardcoded AWS Access Key ID." +severity = "Critical" +confidence = "High" +remediation = "Revoke this key in the AWS IAM console and load credentials from environment variables, an IAM role, or a secrets manager instead." +pattern = '(AKIA[0-9A-Z]{16})' +redact = true + +[[rule]] +id = "SEC-AWS-002" +description = "Hardcoded AWS Secret Access Key." +severity = "Critical" +confidence = "Medium" +remediation = "Revoke this key in the AWS IAM console and load credentials from environment variables, an IAM role, or a secrets manager instead." +pattern = "(?i)aws_secret_access_key\\s*[:=]\\s*['\"]?([A-Za-z0-9/+=]{40})['\"]?" +redact = true + +[[rule]] +id = "SEC-GCP-001" +description = "GCP service account key file detected." +severity = "High" +confidence = "Medium" +remediation = "Do not commit service account key files. Use Workload Identity Federation or a secrets manager instead." +pattern = '"type"\s*:\s*"service_account"' +file_pattern = "*.json" +# No secret value is captured here (just a marker string), but `redact = true` +# is needed anyway so this rule bypasses the Python comment/string-literal +# heuristic: the whole line is a quoted JSON string and would otherwise be +# skipped as a "string literal". +redact = true + +[[rule]] +id = "SEC-GCP-002" +description = "Hardcoded GCP service account private key." +severity = "Critical" +confidence = "High" +remediation = "Revoke this key in the GCP IAM console and remove it from source control. Use Workload Identity Federation or a secrets manager instead." +pattern = '"private_key"\s*:\s*"(-----BEGIN [^"]+?-----[^"]*)"' +file_pattern = "*.json" +redact = true + +[[rule]] +id = "SEC-AZURE-001" +description = "Hardcoded Azure Storage connection string." +severity = "Critical" +confidence = "High" +remediation = "Rotate the storage account key in the Azure Portal and load connection strings from a secrets manager (e.g. Azure Key Vault) instead." +pattern = 'DefaultEndpointsProtocol=https?;AccountName=[^;]+;AccountKey=([A-Za-z0-9+/=]{20,})' +redact = true + +# ------------------------------------------- +# SECTION: SaaS / developer platform tokens +# ------------------------------------------- + +[[rule]] +id = "SEC-GITHUB-001" +description = "Hardcoded GitHub Personal Access Token or OAuth token." +severity = "Critical" +confidence = "High" +remediation = "Revoke this token at https://github.com/settings/tokens and use a secrets manager or CI secret store instead." +pattern = '(gh[pousr]_[A-Za-z0-9]{36,255})' +redact = true + +[[rule]] +id = "SEC-SLACK-001" +description = "Hardcoded Slack token." +severity = "High" +confidence = "High" +remediation = "Revoke this token in your Slack App configuration and load it from a secrets manager instead." +pattern = '(xox[baprs]-[0-9A-Za-z-]{10,72})' +redact = true + +[[rule]] +id = "SEC-STRIPE-001" +description = "Hardcoded Stripe live API key." +severity = "Critical" +confidence = "High" +remediation = "Revoke this key in the Stripe Dashboard and load it from a secrets manager instead." +pattern = '((?:sk|pk|rk)_live_[0-9a-zA-Z]{24,})' +redact = true + +# ------------------------------------------- +# SECTION: Private key material +# ------------------------------------------- + +[[rule]] +id = "SEC-PRIVATEKEY-001" +description = "Private key material (RSA/EC/OpenSSH/DSA/PGP) committed to the repository." +severity = "Critical" +confidence = "High" +remediation = "Remove the private key from source control, rotate the corresponding keypair, and store private keys outside the repository (e.g. a secrets manager or local keychain)." +pattern = '-----BEGIN ((RSA|EC|OPENSSH|DSA|PGP) )?PRIVATE KEY-----' +# redact = true for defense-in-depth (bypasses the comment/string heuristic) +# even though this marker alone reveals only the key type, not key material. +redact = true + +[[rule]] +id = "SEC-JWT-001" +description = "Hardcoded JSON Web Token (JWT)." +severity = "High" +confidence = "Medium" +remediation = "JWTs can carry sensitive claims and may be replayable until expiry. Remove hardcoded tokens and issue them at runtime instead." +pattern = '(eyJ[A-Za-z0-9_-]{5,}\.eyJ[A-Za-z0-9_-]{5,}\.[A-Za-z0-9_-]{10,})' +redact = true + +# ------------------------------------------- +# SECTION: Generic secret assignment +# ------------------------------------------- + +[[rule]] +id = "SEC-GENERIC-001" +description = "Generic hardcoded secret (api_key/secret/password/token assignment)." +severity = "Medium" +confidence = "Low" +remediation = "Load this value from an environment variable or secrets manager instead of hardcoding it." +pattern = "(?i)(?:api[_-]?key|secret|password|passwd|token)\\s*[:=]\\s*['\"]([A-Za-z0-9_./+=-]{8,})['\"]" +redact = true + +# ------------------------------------------- +# SECTION: High-entropy generic secrets +# ------------------------------------------- + +[[entropy_rule]] +id = "SEC-ENTROPY-001" +description = "High-entropy string that may be an undetected secret (base64/hex blob)." +severity = "Medium" +confidence = "Low" +remediation = "Review this value. If it is a credential or key, move it to a secrets manager or environment variable; if it is not sensitive, add it to the baseline." +token_pattern = "[A-Za-z0-9+/_=-]{20,}" +min_length = 20 +threshold = 4.5