use crate::detectors::ast_fingerprint::parse_root_ext;
use crate::detectors::base::{Detector, DetectorConfig};
use crate::detectors::security::ast_helpers::{collect_named_args, node_text, unwrap_callee};
use crate::detectors::security::python_imports::{
class_attr_constructors_of, collect_python_from_imports, collect_python_module_aliases,
PythonAliases,
};
use crate::detectors::security::scan_inputs::{ScanAstInputs, ScanInputs};
use crate::graph::GraphQueryExt;
use crate::models::{Finding, Severity};
use crate::parsers::lightweight::Language;
use anyhow::Result;
use regex::Regex;
use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};
use std::sync::LazyLock;
use tracing::info;
const SUPPORTED_EXTS: &[&str] = &["py", "js", "ts", "jsx", "tsx", "rb", "php", "java", "go"];
const AST_EXTS: &[&str] = &["py"];
static COOKIE_PATTERN: LazyLock<Regex> = LazyLock::new(|| {
Regex::new(
r"(?i)(\.set_cookie\s*\(|response\.set_cookie\s*\(|res\.cookie\s*\(|response\.cookie\s*\(|setcookie\s*\()",
)
.expect("valid regex")
});
pub struct InsecureCookieDetector {
repository_path: PathBuf,
max_findings: usize,
}
impl InsecureCookieDetector {
crate::detectors::detector_new!(50);
fn relative_path(&self, path: &Path) -> PathBuf {
crate::detectors::detector_relative_path(&self.repository_path, path)
}
fn is_sensitive_cookie(line: &str, surrounding: &str) -> (bool, String) {
let combined = format!("{} {}", line, surrounding).to_lowercase();
if combined.contains("session") {
return (true, "session".to_string());
}
if combined.contains("auth") || combined.contains("token") {
return (true, "authentication".to_string());
}
if combined.contains("jwt") || combined.contains("bearer") {
return (true, "JWT/bearer token".to_string());
}
if combined.contains("csrf") || combined.contains("xsrf") {
return (true, "CSRF token".to_string());
}
if combined.contains("remember") || combined.contains("login") {
return (true, "remember-me/login".to_string());
}
if combined.contains("user") || combined.contains("account") {
return (true, "user data".to_string());
}
(false, "general".to_string())
}
fn check_cookie_flags(lines: &[&str], cookie_line: usize) -> CookieFlags {
let start = cookie_line.saturating_sub(3);
let end = (cookie_line + 15).min(lines.len());
let context = lines[start..end].join(" ").to_lowercase();
CookieFlags {
has_httponly: context.contains("httponly"),
has_secure: context.contains("secure") && !context.contains("insecure"),
has_samesite: context.contains("samesite"),
samesite_value: if context.contains("samesite=strict")
|| context.contains("samesite='strict'")
{
Some("Strict".to_string())
} else if context.contains("samesite=lax") || context.contains("samesite='lax'") {
Some("Lax".to_string())
} else if context.contains("samesite=none") || context.contains("samesite='none'") {
Some("None".to_string())
} else {
None
},
}
}
#[allow(clippy::too_many_arguments)]
fn build_finding_from_site(
&self,
path: &Path,
line_num: u32,
line_text: &str,
surrounding: &str,
flags: &CookieFlags,
ext: &str,
graph: &dyn crate::graph::GraphQuery,
) -> Option<Finding> {
let (is_sensitive, cookie_type) = Self::is_sensitive_cookie(line_text, surrounding);
let mut missing = Vec::new();
if !flags.has_httponly {
missing.push("HttpOnly");
}
if !flags.has_secure {
missing.push("Secure");
}
if !flags.has_samesite {
missing.push("SameSite");
}
if missing.is_empty() {
return None;
}
let severity = if is_sensitive && !flags.has_httponly {
Severity::Critical
} else if is_sensitive {
Severity::High
} else if !flags.has_httponly {
Severity::Medium
} else {
Severity::Low
};
let path_str = path.to_string_lossy().to_string();
let containing_func = graph
.find_function_at(&path_str, line_num)
.map(|f| f.node_name(graph.interner()).to_string());
let mut notes = Vec::new();
if is_sensitive {
notes.push(format!("🔐 {} cookie - high value target", cookie_type));
}
notes.push(format!("❌ Missing: {}", missing.join(", ")));
if let Some(ss) = &flags.samesite_value {
notes.push(format!("✓ SameSite={}", ss));
if ss == "None" && !flags.has_secure {
notes.push("⚠️ SameSite=None requires Secure flag!".to_string());
}
}
if let Some(func) = containing_func {
notes.push(format!("📦 In function: `{}`", func));
}
let context_notes = format!("\n\n**Analysis:**\n{}", notes.join("\n"));
let suggestion = match ext {
"py" => "```python\n\
response.set_cookie(\n\
'cookie_name',\n\
value,\n\
httponly=True, # Prevents JavaScript access\n\
secure=True, # HTTPS only\n\
samesite='Lax' # CSRF protection\n\
)\n\
```"
.to_string(),
"js" | "ts" => "```javascript\n\
res.cookie('cookie_name', value, {\n\
httpOnly: true, // Prevents JavaScript access\n\
secure: true, // HTTPS only\n\
sameSite: 'lax' // CSRF protection\n\
});\n\
```"
.to_string(),
"php" => "```php\n\
setcookie('cookie_name', $value, [\n\
'httponly' => true,\n\
'secure' => true,\n\
'samesite' => 'Lax'\n\
]);\n\
```"
.to_string(),
_ => "Add httponly, secure, and samesite flags.".to_string(),
};
Some(Finding {
id: String::new(),
detector: "InsecureCookieDetector".to_string(),
severity,
title: format!(
"Cookie missing {} flag{}",
missing[0],
if missing.len() > 1 { "s" } else { "" }
),
description: format!(
"Cookie is missing security flags that protect against common attacks.{}",
context_notes
),
affected_files: vec![self.relative_path(path)],
line_start: Some(line_num),
line_end: Some(line_num),
suggested_fix: Some(suggestion),
estimated_effort: Some("5 minutes".to_string()),
category: Some("security".to_string()),
cwe_id: Some(if !flags.has_httponly {
"CWE-1004".to_string()
} else {
"CWE-614".to_string()
}),
why_it_matters: Some(
"• **HttpOnly** prevents XSS attacks from stealing cookies via JavaScript\n\
• **Secure** ensures cookies are only sent over HTTPS\n\
• **SameSite** prevents CSRF attacks by controlling cross-site requests"
.to_string(),
),
..Default::default()
})
}
fn scan_file_ast(
&self,
inputs: &ScanAstInputs<'_>,
graph: &dyn crate::graph::GraphQuery,
) -> Vec<Finding> {
let path = inputs.path();
let content = inputs.content();
let ext = inputs.ext();
let lang = inputs.lang;
let cached_tree = inputs.cached_tree;
let mut findings = vec![];
if content.contains('\0') || content.len() > 500_000 {
return findings;
}
let owned;
let root = match cached_tree {
Some(tree) => tree.root_node(),
None => match parse_root_ext(content, lang, ext) {
Some(t) => {
owned = t;
owned.root_node()
}
None => return findings,
},
};
let bytes = content.as_bytes();
let lines: Vec<&str> = content.lines().collect();
let alias_map = collect_python_from_imports(root, bytes);
let module_aliases = collect_python_module_aliases(root, bytes);
let aliases = PythonAliases::new(&alias_map, &module_aliases);
let jar_ctor_names: HashSet<&str> = HashSet::from([
"CookieJar",
"RequestsCookieJar",
"LWPCookieJar",
"MozillaCookieJar",
"FileCookieJar",
"http.cookiejar.CookieJar",
"requests.cookies.RequestsCookieJar",
"aiohttp.CookieJar",
]);
let class_jar_attrs = class_attr_constructors_of(root, bytes, &aliases, &jar_ctor_names);
let mut sites: Vec<CookieSite<'_>> = Vec::new();
collect_cookie_sites(root, bytes, &class_jar_attrs, &mut sites);
for site in sites {
if findings.len() >= self.max_findings {
break;
}
let line_idx = site.line;
if let Some(line) = lines.get(line_idx) {
let prev = if line_idx > 0 {
Some(lines[line_idx - 1])
} else {
None
};
if crate::detectors::is_line_suppressed(line, prev) {
continue;
}
}
let line_text = lines.get(line_idx).copied().unwrap_or("");
let start = line_idx.saturating_sub(5);
let end = (line_idx + 5).min(lines.len());
let surrounding = lines[start..end].join(" ");
let flags = Self::check_cookie_flags(&lines, line_idx);
let line_num = (line_idx + 1) as u32;
if let Some(finding) = self.build_finding_from_site(
path,
line_num,
line_text,
&surrounding,
&flags,
ext,
graph,
) {
findings.push(finding);
}
}
findings
}
fn scan_file_line(
&self,
inputs: &ScanInputs<'_>,
graph: &dyn crate::graph::GraphQuery,
) -> Vec<Finding> {
let path = inputs.path;
let content = inputs.content;
let ext = inputs.ext;
let mut findings = vec![];
if content.len() > 500_000 {
return findings;
}
let lines: Vec<&str> = content.lines().collect();
for (i, line) in lines.iter().enumerate() {
if findings.len() >= self.max_findings {
break;
}
let prev = if i > 0 { Some(lines[i - 1]) } else { None };
if crate::detectors::is_line_suppressed(line, prev) {
continue;
}
if !COOKIE_PATTERN.is_match(line) {
continue;
}
let start = i.saturating_sub(5);
let end = (i + 5).min(lines.len());
let surrounding = lines[start..end].join(" ");
let flags = Self::check_cookie_flags(&lines, i);
let line_num = (i + 1) as u32;
if let Some(finding) =
self.build_finding_from_site(path, line_num, line, &surrounding, &flags, ext, graph)
{
findings.push(finding);
}
}
findings
}
}
struct CookieFlags {
has_httponly: bool,
has_secure: bool,
has_samesite: bool,
samesite_value: Option<String>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum CookieSiteKind {
ServerSetCookie,
}
struct CookieSite<'a> {
#[allow(dead_code)] call_node: tree_sitter::Node<'a>,
#[allow(dead_code)]
kind: CookieSiteKind,
line: usize,
}
fn collect_cookie_sites<'a>(
node: tree_sitter::Node<'a>,
source: &[u8],
class_jar_attrs: &HashMap<String, HashSet<String>>,
out: &mut Vec<CookieSite<'a>>,
) {
if node.kind() == "call" {
if let Some(site) = classify_cookie_call(node, source, class_jar_attrs) {
out.push(site);
}
}
let mut c = node.walk();
for child in node.children(&mut c) {
collect_cookie_sites(child, source, class_jar_attrs, out);
}
}
const SHAPE_CLASSIFIED_METHODS: &[&str] = &["set_cookie", "set_signed_cookie"];
const STDLIB_ONLY_METHODS: &[&str] = &["add_cookie_header", "extract_cookies"];
fn classify_cookie_call<'a>(
call: tree_sitter::Node<'a>,
source: &[u8],
class_jar_attrs: &HashMap<String, HashSet<String>>,
) -> Option<CookieSite<'a>> {
let func = call.child_by_field_name("function")?;
let func = unwrap_callee(func);
if func.kind() != "attribute" {
return None;
}
let attr_node = func.child_by_field_name("attribute")?;
let receiver = func.child_by_field_name("object")?;
let method_name = node_text(attr_node, source)?;
if STDLIB_ONLY_METHODS.contains(&method_name) {
return None;
}
if !SHAPE_CLASSIFIED_METHODS.contains(&method_name) {
return None;
}
if let Some(attr) = self_attr_text(receiver, source) {
if let Some(class_name) = enclosing_class_name(call, source) {
if let Some(jar_attrs) = class_jar_attrs.get(&class_name) {
if jar_attrs.contains(&attr) {
return None;
}
}
}
}
let args_node = call.child_by_field_name("arguments")?;
let arg_nodes = collect_named_args(args_node);
let mut positional = 0usize;
let mut kwargs = 0usize;
for a in &arg_nodes {
if a.kind() == "keyword_argument" {
kwargs += 1;
} else {
positional += 1;
}
}
if positional <= 1 && kwargs == 0 {
return None;
}
Some(CookieSite {
call_node: call,
kind: CookieSiteKind::ServerSetCookie,
line: call.start_position().row,
})
}
fn self_attr_text(node: tree_sitter::Node<'_>, source: &[u8]) -> Option<String> {
if node.kind() != "attribute" {
return None;
}
let obj = node.child_by_field_name("object")?;
if node_text(obj, source)? != "self" {
return None;
}
let attr = node.child_by_field_name("attribute")?;
Some(node_text(attr, source)?.to_string())
}
fn enclosing_class_name(node: tree_sitter::Node<'_>, source: &[u8]) -> Option<String> {
let mut cur = node.parent();
while let Some(p) = cur {
if p.kind() == "class_definition" {
let name = p.child_by_field_name("name")?;
return Some(node_text(name, source)?.to_string());
}
cur = p.parent();
}
None
}
impl Detector for InsecureCookieDetector {
fn name(&self) -> &'static str {
"insecure-cookie"
}
fn description(&self) -> &'static str {
"Detects cookies without security flags"
}
fn bypass_postprocessor(&self) -> bool {
true
}
fn requires_graph(&self) -> bool {
false
}
fn file_extensions(&self) -> &'static [&'static str] {
SUPPORTED_EXTS
}
fn detect(
&self,
ctx: &crate::detectors::analysis_context::AnalysisContext,
) -> Result<Vec<Finding>> {
let graph = ctx.graph;
let files = &ctx.as_file_provider();
let mut findings: Vec<Finding> = Vec::new();
for path in files.files_with_extensions(SUPPORTED_EXTS) {
if findings.len() >= self.max_findings {
break;
}
let path_str = path.to_string_lossy().to_string();
if crate::detectors::base::is_test_path(&path_str) {
continue;
}
let raw = match files.content(path) {
Some(c) => c,
None => continue,
};
if !raw.contains("set_cookie")
&& !raw.contains("set_signed_cookie")
&& !raw.contains("setCookie")
&& !raw.contains("setcookie")
&& !raw.contains(".cookie(")
&& !raw.contains("res.cookie")
{
continue;
}
let ext = path.extension().and_then(|e| e.to_str()).unwrap_or("");
let new_findings = if AST_EXTS.contains(&ext) {
let masked = match files.masked_content(path) {
Some(m) => m,
None => continue,
};
let cached = files.tree(path);
let lang = Language::from_path(path);
let scan = ScanInputs::new(path, &masked, ext);
let ast_inputs = ScanAstInputs::new(scan, lang, cached.as_deref());
self.scan_file_ast(&ast_inputs, graph)
} else {
let masked = match files.masked_content(path) {
Some(m) => m,
None => continue,
};
let scan = ScanInputs::new(path, &masked, ext);
self.scan_file_line(&scan, graph)
};
findings.extend(new_findings);
}
info!(
"InsecureCookieDetector found {} findings (AST-aware Python, regex fallback for JS/PHP/...)",
findings.len()
);
Ok(findings)
}
}
impl crate::detectors::RegisteredDetector for InsecureCookieDetector {
fn create(init: &crate::detectors::DetectorInit) -> std::sync::Arc<dyn Detector> {
std::sync::Arc::new(Self::new(init.repo_path))
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::graph::builder::GraphBuilder;
fn run(file: &str, content: &str) -> Vec<Finding> {
let store = GraphBuilder::new().freeze();
let detector = InsecureCookieDetector::new("/mock/repo");
let ctx = crate::detectors::analysis_context::AnalysisContext::test_with_mock_files(
&store,
vec![(file, content)],
);
detector.detect(&ctx).expect("detection should succeed")
}
#[test]
fn test_detects_insecure_cookie() {
let findings = run(
"app.py",
"from flask import make_response\n\
\n\
def set_session(user_id):\n\
\x20 resp = make_response(\"OK\")\n\
\x20 resp.set_cookie('session_id', user_id)\n\
\x20 return resp\n",
);
assert!(
!findings.is_empty(),
"Should detect cookie without security flags"
);
assert!(
findings.iter().any(|f| f.title.contains("HttpOnly")
|| f.title.contains("Secure")
|| f.title.contains("SameSite")),
"Finding should mention missing flag. Titles: {:?}",
findings.iter().map(|f| &f.title).collect::<Vec<_>>()
);
}
#[test]
fn test_no_finding_for_secure_cookie() {
let findings = run(
"app.py",
"from flask import make_response\n\
\n\
def set_session(user_id):\n\
\x20 resp = make_response(\"OK\")\n\
\x20 resp.set_cookie('session_id', user_id, httponly=True, secure=True, samesite='Lax')\n\
\x20 return resp\n",
);
assert!(
findings.is_empty(),
"Should not detect anything for secure cookie. Found: {:?}",
findings.iter().map(|f| &f.title).collect::<Vec<_>>()
);
}
#[test]
fn test_no_finding_for_enum_cookie_value() {
let findings = run(
"params.py",
"from enum import Enum\n\
\n\
class ParamTypes(Enum):\n\
\x20 query = \"query\"\n\
\x20 header = \"header\"\n\
\x20 cookie = \"cookie\"\n",
);
assert!(
findings.is_empty(),
"Should not flag enum values containing 'cookie'. Found: {:?}",
findings.iter().map(|f| &f.title).collect::<Vec<_>>()
);
}
#[test]
fn test_no_finding_for_cookie_class_field() {
let findings = run(
"models.py",
"class SecurityScheme:\n\
\x20 cookie = \"apiKeyCookie\"\n\
\x20 header = \"apiKeyHeader\"\n",
);
assert!(
findings.is_empty(),
"Should not flag class field assignments. Found: {:?}",
findings.iter().map(|f| &f.title).collect::<Vec<_>>()
);
}
#[test]
fn test_no_finding_for_cookie_attribute_access() {
let findings = run(
"response.py",
"def set_cookie(self, key, value):\n\
\x20 self.cookies[key] = value\n\
\x20 self.cookies[key][\"secure\"] = True\n\
\x20 self.cookies[key][\"httponly\"] = True\n",
);
assert!(
findings.is_empty(),
"Should not flag self.cookies[] attribute access. Found: {:?}",
findings.iter().map(|f| &f.title).collect::<Vec<_>>()
);
}
#[test]
fn test_no_finding_for_multiline_set_cookie_with_flags() {
let findings = run(
"middleware.py",
"def process_response(self, request, response):\n\
\x20 response.set_cookie(\n\
\x20 settings.SESSION_COOKIE_NAME,\n\
\x20 request.session.session_key,\n\
\x20 max_age=max_age,\n\
\x20 expires=expires,\n\
\x20 domain=settings.SESSION_COOKIE_DOMAIN,\n\
\x20 path=settings.SESSION_COOKIE_PATH,\n\
\x20 secure=settings.SESSION_COOKIE_SECURE or None,\n\
\x20 httponly=settings.SESSION_COOKIE_HTTPONLY or None,\n\
\x20 samesite=settings.SESSION_COOKIE_SAMESITE,\n\
\x20 )\n",
);
assert!(
findings.is_empty(),
"Should detect flags in multi-line set_cookie() call. Found: {:?}",
findings.iter().map(|f| &f.title).collect::<Vec<_>>()
);
}
#[test]
fn test_no_finding_for_jar_set_cookie_single_arg() {
let findings = run(
"client.py",
"from http.cookiejar import CookieJar\n\
\n\
def store(jar, cookie):\n\
\x20 jar.set_cookie(cookie)\n",
);
assert!(
findings.is_empty(),
"Single-arg jar.set_cookie must not fire. Got: {:?}",
findings.iter().map(|f| &f.title).collect::<Vec<_>>()
);
}
#[test]
fn test_no_finding_for_self_jar_set_cookie_in_cookies_class() {
let findings = run(
"_models.py",
"from http.cookiejar import CookieJar\n\
\n\
class Cookies:\n\
\x20 def __init__(self):\n\
\x20 self.jar = CookieJar()\n\
\x20 def set(self, cookie):\n\
\x20 self.jar.set_cookie(cookie)\n",
);
assert!(
findings.is_empty(),
"self.jar.set_cookie inside Cookies-with-self.jar=CookieJar() must not fire. \
This is the httpx FP. Got: {:?}",
findings.iter().map(|f| &f.title).collect::<Vec<_>>()
);
}
#[test]
fn test_no_finding_for_add_cookie_header() {
let findings = run(
"client.py",
"def f(jar, req):\n\
\x20 jar.add_cookie_header(req)\n",
);
assert!(
findings.is_empty(),
"add_cookie_header must not fire. Got: {:?}",
findings.iter().map(|f| &f.title).collect::<Vec<_>>()
);
}
#[test]
fn test_no_finding_for_extract_cookies() {
let findings = run(
"client.py",
"def f(jar, resp, req):\n\
\x20 jar.extract_cookies(resp, req)\n",
);
assert!(
findings.is_empty(),
"extract_cookies must not fire. Got: {:?}",
findings.iter().map(|f| &f.title).collect::<Vec<_>>()
);
}
#[test]
fn test_finding_for_flask_set_cookie_string_kwargs() {
let findings = run(
"app.py",
"def view(response, value):\n\
\x20 response.set_cookie('sid', value)\n",
);
assert!(
!findings.is_empty(),
"Flask 2-arg response.set_cookie must fire. Got: {:?}",
findings.iter().map(|f| &f.title).collect::<Vec<_>>()
);
}
#[test]
fn test_finding_for_flask_set_cookie_with_kwargs() {
let findings = run(
"app.py",
"def view(response, value):\n\
\x20 response.set_cookie('sid', value, httponly=True, secure=True, samesite='Lax')\n",
);
assert!(
findings.is_empty(),
"set_cookie with all flags must not fire. Got: {:?}",
findings.iter().map(|f| &f.title).collect::<Vec<_>>()
);
}
#[test]
fn test_finding_for_flask_set_cookie_variable_args() {
let findings = run(
"app.py",
"def view(response, key, value):\n\
\x20 response.set_cookie(key, value)\n",
);
assert!(
!findings.is_empty(),
"set_cookie(key, value) (2 variables) must fire. Got: {:?}",
findings.iter().map(|f| &f.title).collect::<Vec<_>>()
);
}
#[test]
fn test_no_finding_for_jar_set_cookie_with_subscript() {
let findings = run(
"client.py",
"def f(jar, cookies):\n\
\x20 jar.set_cookie(cookies[0])\n",
);
assert!(
findings.is_empty(),
"1-arg jar.set_cookie(cookies[0]) must not fire. Got: {:?}",
findings.iter().map(|f| &f.title).collect::<Vec<_>>()
);
}
#[test]
fn test_no_finding_for_jar_set_cookie_with_attr() {
let findings = run(
"client.py",
"class Wrapper:\n\
\x20 def store(self):\n\
\x20 self.jar.set_cookie(self.cookie)\n",
);
assert!(
findings.is_empty(),
"self.jar.set_cookie(self.cookie) must not fire. Got: {:?}",
findings.iter().map(|f| &f.title).collect::<Vec<_>>()
);
}
#[test]
fn test_finding_for_django_set_signed_cookie() {
let findings = run(
"views.py",
"def view(response, value):\n\
\x20 response.set_signed_cookie('sid', value)\n",
);
assert!(
!findings.is_empty(),
"Django set_signed_cookie must fire. Got: {:?}",
findings.iter().map(|f| &f.title).collect::<Vec<_>>()
);
}
}