use crate::{AttackCategory, DetectionResult, Detector, Severity, regex_detect};
use regex::Regex;
use std::sync::LazyLock;
static STRONG_PATTERNS: LazyLock<Regex> = LazyLock::new(|| {
Regex::new(
&[
r#"(?i)__proto__(?:["'\[\]])"#,
r"|__proto__\s*=[ \t]*[^=\s]",
r"|constructor\[",
r"|constructor\.prototype",
r"|__defineGetter__",
r"|__defineSetter__",
r"|__lookupGetter__",
r"|__lookupSetter__",
r"|hasOwnProperty\[",
]
.concat(),
)
.unwrap()
});
static WEAK_PATTERNS: LazyLock<Regex> = LazyLock::new(|| Regex::new(r"(?i)__proto__").unwrap());
pub struct PrototypePollutionDetector;
impl Detector for PrototypePollutionDetector {
fn name(&self) -> &'static str {
"prototype_pollution"
}
fn detect(&self, input: &str) -> Option<DetectionResult> {
regex_detect(
std::slice::from_ref(&*STRONG_PATTERNS),
self.name(),
AttackCategory::Data,
Severity::High,
"JavaScript prototype pollution detected",
input,
)
.or_else(|| {
regex_detect(
std::slice::from_ref(&*WEAK_PATTERNS),
self.name(),
AttackCategory::Data,
Severity::Low,
"__proto__ referenced (weak signal)",
input,
)
})
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn name_returns_attack_type() {
assert_eq!(PrototypePollutionDetector.name(), "prototype_pollution");
}
#[test]
fn detects_proto_and_constructor_payloads() {
for payload in [
r#"{"__proto__": {"isAdmin": true}}"#,
r#"{"__proto__": {"polluted": true}}"#,
"obj.constructor.prototype.isAdmin = true",
"a[constructor[0]]",
"o[__proto__][isAdmin]",
] {
let r = PrototypePollutionDetector
.detect(payload)
.unwrap_or_else(|| panic!("expected detection for {:?}", payload));
assert_eq!(r.attack_type, "prototype_pollution");
assert_eq!(r.category, AttackCategory::Data);
assert_eq!(r.severity, Severity::High);
assert!(
!r.matched_pattern.is_empty(),
"matched_pattern empty for {:?}",
payload
);
assert!(
r.offset <= payload.len(),
"offset out of range for {:?}",
payload
);
}
}
#[test]
fn detects_legacy_getter_setter_apis() {
for payload in [
"__defineGetter__('x', fn)",
"__defineSetter__('x', fn)",
"__lookupGetter__('x')",
"__lookupSetter__('x')",
"hasOwnProperty['isAdmin']",
] {
let r = PrototypePollutionDetector
.detect(payload)
.unwrap_or_else(|| panic!("expected detection for {:?}", payload));
assert!(
!r.matched_pattern.is_empty(),
"matched_pattern empty for {:?}",
payload
);
assert!(
r.offset <= payload.len(),
"offset out of range for {:?}",
payload
);
}
}
#[test]
fn bare_proto_reference_is_low_key_shape_is_high() {
for payload in [
r#"{"__proto__": {"isAdmin": true}}"#,
r#"{"__proto__":{"polluted":1}}"#,
"o[__proto__][isAdmin]",
"obj.__proto__ = {}",
"?__proto__[x]=1",
] {
let r = PrototypePollutionDetector
.detect(payload)
.unwrap_or_else(|| panic!("expected detection for {:?}", payload));
assert_eq!(r.severity, Severity::High, "payload {:?}", payload);
}
for input in [
"const p = obj.__proto__;",
"if (obj.__proto__ === Array.prototype) { init(); }",
"| **prototype_pollution** | `__proto__` 原型链污染 | High |",
] {
let r = PrototypePollutionDetector
.detect(input)
.unwrap_or_else(|| panic!("expected weak detection for {:?}", input));
assert_eq!(r.severity, Severity::Low, "input {:?}", input);
}
}
#[test]
fn ignores_benign_inputs() {
for input in [
"Hello, this is a normal text input.",
"constructor",
"hasOwnProperty",
"proto",
"the prototype chain is a concept",
] {
assert!(
PrototypePollutionDetector.detect(input).is_none(),
"false positive: {:?}",
input
);
}
}
#[test]
fn edge_cases() {
assert!(PrototypePollutionDetector.detect("").is_none());
assert!(PrototypePollutionDetector.detect(" ").is_none());
assert!(PrototypePollutionDetector.detect("__proto__").is_none()); assert!(PrototypePollutionDetector.detect("Прототип").is_none());
}
}