use crate::{AttackCategory, DetectionResult, Detector, Severity, regex_detect};
use regex::Regex;
use std::sync::LazyLock;
static STRONG_PATTERNS: LazyLock<Regex> = LazyLock::new(|| {
Regex::new(concat!(
r#"\{\{[^{}]{0,120}?\d+[ \t]*[-+*/%][ \t]*['"]?\d+[^{}]{0,120}?\}\}"#,
r#"|\$\{[^{}]{0,120}?\d+[ \t]*[-+*/%][ \t]*['"]?\d+[^{}]{0,120}?\}"#,
r#"|<%=[^{}]{0,120}?\d+[ \t]*[-+*/%][ \t]*['"]?\d+[^{}]{0,120}?%>"#,
r"|\{\{[ \t]*config\b",
r"|\$\{[ \t]*(?:T[ \t]*\(|@[\w.]+@)",
r#"|\{%\s*(?:include|extends|import|from)\s+['"]\s*(?:/|\.\.)"#,
r"|\{\{[^{}%]{0,160}?__(?:class|dict|mro|subclasses|globals|builtins)__|\{%[^{}%]{0,160}?__(?:class|dict|mro|subclasses|globals|builtins)__|\$\{[^{}%]{0,160}?__(?:class|dict|mro|subclasses|globals|builtins)__",
r"|(?i:\?new\s*\()",
))
.unwrap()
});
static WEAK_PATTERNS: LazyLock<Regex> = LazyLock::new(|| {
Regex::new(concat!(
r"\{%\s*.*?\s*%\}",
r"|<%=",
r"|<%@",
r"|#set\s*\(",
r"|(?i:<#assign\s)",
r"|__(?:class|dict|mro|subclasses|globals|builtins)__",
))
.unwrap()
});
pub struct SstiDetector;
impl Detector for SstiDetector {
fn name(&self) -> &'static str {
"ssti"
}
fn detect(&self, input: &str) -> Option<DetectionResult> {
regex_detect(
std::slice::from_ref(&*STRONG_PATTERNS),
self.name(),
AttackCategory::Injection,
Severity::Critical,
"Server-Side Template Injection detected",
input,
)
.or_else(|| {
regex_detect(
std::slice::from_ref(&*WEAK_PATTERNS),
self.name(),
AttackCategory::Injection,
Severity::Low,
"Template directive / magic attribute present (weak signal)",
input,
)
})
}
}
#[cfg(test)]
mod tests {
use super::*;
fn det() -> SstiDetector {
SstiDetector
}
fn assert_hit(input: &str) {
assert_hit_at(input, Severity::Critical);
}
fn assert_hit_at(input: &str, severity: Severity) {
crate::test_helpers::assert_detected(&det(), input, AttackCategory::Injection, severity);
}
#[test]
fn name_is_ssti() {
assert_eq!(det().name(), "ssti");
}
#[test]
fn detects_common_payloads() {
for (input, severity) in [
("{{7*7}}", Severity::Critical),
("{{ ''.__class__.__mro__[1].__subclasses__() }}", Severity::Critical),
("${7*7}", Severity::Critical),
("{% include '/etc/passwd' %}", Severity::Critical),
("{{config.__init__.__globals__}}", Severity::Critical),
("<%= params[:x] %>", Severity::Low),
(r#"<%@ page import="java.util.*" %>"#, Severity::Low),
("#set($x = 5)", Severity::Low),
] {
assert_hit_at(input, severity);
}
}
#[test]
fn template_source_is_low_not_critical() {
for input in [
"{% if user.is_admin %}\n <p>Admin</p>\n{% endif %}",
"{% for item in items %}\n <li>{{ item.name }}</li>\n{% endfor %}",
"<h1>Hello <%= @user.name %></h1>",
r#"<%@ page language="java" contentType="text/html; charset=UTF-8" %>"#,
"#set($x = 5)",
"<#assign title=\"Home\">",
] {
assert_hit_at(input, Severity::Low);
}
}
#[test]
fn bare_python_dunders_are_low_but_delimited_ones_stay_critical() {
for input in [
"return f\"<Store {self.__class__.__name__}>\"",
"return dict(self.__dict__)",
"return [c for c in Plugin.__subclasses__()]",
] {
assert_hit_at(input, Severity::Low);
}
for input in [
"{{ self.__dict__ }}",
"{{ ''.__class__.__mro__ }}",
"${config.__class__}",
] {
assert_hit(input);
}
}
#[test]
fn python_protocol_methods_not_detected() {
for input in [
"def __getitem__(self, key):\n return self._data[key]",
"__set_name__ is called by the descriptor protocol.",
"def __getattr__(self, name):\n raise AttributeError(name)",
] {
crate::test_helpers::assert_clean(&det(), input);
}
}
#[test]
fn benign_inputs_not_detected() {
for input in [
"Hello, this is a normal text input. Nothing suspicious here.",
"The total is $5.00 plus tax",
"Please enter your name below",
"The class of 2026 graduates in May",
"100% of users agree with this",
] {
assert!(det().detect(input).is_none(), "false positive: {input}");
}
}
#[test]
fn plain_interpolation_not_detected() {
for input in [
"the price is ${amount}",
"${user}${pass}",
"@Value(\"${x.y.z}\")",
"${env:JAVA_HOME}",
"const x = `${name}`",
"${#strings.toUpperCase(name)}",
"{{ name }}",
"{{ app_config }}",
"${timeout:30s}",
"${PATH:-/usr/bin}",
] {
assert!(det().detect(input).is_none(), "false positive: {input}");
}
}
#[test]
fn evaluation_probes_and_runtime_access_detected() {
for input in [
"${7*7}",
"{{7*7}}",
"{{ 7 * 7 }}",
"{{ 10 % 2 }}",
"total = {{ 10 % 2 }}",
"${{7*7}}",
"<%= 7*7 %>",
"{{config}}",
"{{ config.items }}",
"${T(java.lang.Runtime)}",
"${@java.lang.Runtime@getRuntime()}",
"{{ ''.__class__.__mro__ }}",
"{{ self.__dict__ }}",
] {
assert_hit(input);
}
}
#[test]
fn freemarker_and_quoted_operand_detected() {
for input in [
"{{7*'7'}}",
r#"{{7*"7"}}"#,
"<#assign ex=\"freemarker.template.utility.Execute\"?new()>${ex(\"id\")}",
r#"${"freemarker.template.utility.Execute"?new()("id")}"#,
] {
assert_hit(input);
}
}
#[test]
fn assign_and_new_lookalikes_not_detected() {
for input in [
r##"<a href="#assign something">x</a>"##,
r##"<a href="#assign">x</a>"##,
"https://example.com/#assign-section",
"q?new=1", "{{ '7' }}", r#"{{ "hello" * 3 }}"#, "{{ count % 2 }}",
"{{ n % 10 }}",
"{{ a % b }}",
"The template renders the remainder of {{ a % b }} on the page.",
] {
assert!(det().detect(input).is_none(), "false positive: {input}");
}
}
#[test]
fn edge_cases() {
assert!(det().detect("").is_none());
assert!(det().detect(" \t\n ").is_none());
assert!(det().detect("你好世界 こんにちは").is_none());
assert!(det().detect("{7*7}").is_none());
assert!(det().detect("{{7*7").is_none());
assert!(det().detect("__CLASS__").is_none());
assert!(det().detect("{$x=7}").is_none());
}
#[test]
fn obfuscated_variants_detected() {
for input in [
"{{ ''.__class__.__MRO__[1] }}",
"{{ self.__dict__ }}",
"{{request.application.__globals__}}",
] {
assert_hit(input);
}
}
}