import requests
class SecurityQuestionTester:
def __init__(self, base_url):
self.base_url = base_url
self.session = requests.Session()
COMMON_ANSWERS = {
"color": ["blue", "red", "green", "black", "purple", "pink", "white", "yellow"],
"pet": ["max", "buddy", "charlie", "lucy", "bella", "daisy", "rocky", "jack", "bear", "duke"],
"city": ["new york", "los angeles", "chicago", "houston", "philadelphia", "phoenix", "san diego"],
"team": ["yankees", "cowboys", "lakers", "patriots", "giants", "bears", "packers"],
"movie": ["star wars", "godfather", "avatar", "titanic", "matrix", "inception"],
"food": ["pizza", "steak", "pasta", "sushi", "tacos", "burgers", "chicken"],
"car": ["ford", "toyota", "honda", "bmw", "mercedes", "chevrolet", "tesla"],
}
def get_questions(self, email=None):
"""Get available security questions"""
url = f"{self.base_url}/api/security-questions"
if email:
url += f"?email={email}"
response = self.session.get(url)
return response.json() if response.status_code == 200 else None
def analyze_question(self, question):
"""Analyze question strength"""
question_lower = question.lower()
weakness = {
"question": question,
"risk": "unknown",
"issues": []
}
weak_patterns = {
"maiden name": ("High", "Public records searchable"),
"pet": ("High", "Often on social media"),
"born": ("High", "Public records"),
"school": ("High", "LinkedIn/Facebook"),
"color": ("High", "Very limited options"),
"team": ("High", "Social media posts"),
"first car": ("Medium", "May be shared publicly"),
"best friend": ("Medium", "Social media"),
"movie": ("Medium", "Limited popular options"),
}
for pattern, (risk, reason) in weak_patterns.items():
if pattern in question_lower:
weakness["risk"] = risk
weakness["issues"].append(reason)
if not weakness["issues"]:
weakness["risk"] = "Unknown"
weakness["issues"].append("Requires manual analysis")
return weakness
def brute_force_answer(self, email, answer_type, max_attempts=50):
"""Attempt to guess security answer"""
answers = self.COMMON_ANSWERS.get(answer_type, [])
for i, answer in enumerate(answers[:max_attempts]):
response = self.session.post(
f"{self.base_url}/api/verify-security-answer",
json={"email": email, "answer": answer}
)
if response.status_code == 429:
return {"status": "rate_limited", "attempts": i + 1}
if "success" in response.text.lower() or "token" in response.text.lower():
return {"status": "success", "answer": answer, "attempts": i + 1}
return {"status": "failed", "attempts": len(answers)}
def test_rate_limiting(self, email, num_attempts=100):
"""Test rate limiting on answer attempts"""
for i in range(num_attempts):
response = self.session.post(
f"{self.base_url}/api/verify-security-answer",
json={"email": email, "answer": f"wrong{i}"}
)
if response.status_code == 429:
return {"rate_limited": True, "after_attempts": i + 1}
return {"rate_limited": False, "attempts_tested": num_attempts}
def generate_report(self, questions):
"""Generate security question analysis report"""
print("\n=== SECURITY QUESTION ANALYSIS ===\n")
if not questions:
print("No questions retrieved")
return
for q in questions:
analysis = self.analyze_question(q.get("question", q))
print(f"Question: {analysis['question']}")
print(f" Risk Level: {analysis['risk']}")
for issue in analysis['issues']:
print(f" Issue: {issue}")
print()
tester = SecurityQuestionTester("https://target.com")
questions = tester.get_questions()
tester.generate_report(questions)
print(tester.test_rate_limiting("test@example.com"))
class SecureSecurityQuestions:
MINIMUM_ANSWER_LENGTH = 4
MAX_ATTEMPTS = 3
LOCKOUT_DURATION = 3600
def __init__(self, redis_client):
self.redis = redis_client
def validate_answer_format(self, answer):
"""Validate answer meets minimum requirements"""
if len(answer) < self.MINIMUM_ANSWER_LENGTH:
return False, f"Answer must be at least {self.MINIMUM_ANSWER_LENGTH} characters"
return True, None
def verify_answer(self, user_id, provided_answer):
"""Verify answer with rate limiting"""
attempt_key = f"security_question_attempts:{user_id}"
if self.redis.exists(f"security_question_locked:{user_id}"):
return False, "Account temporarily locked"
attempts = self.redis.incr(attempt_key)
if attempts == 1:
self.redis.expire(attempt_key, 3600)
if attempts > self.MAX_ATTEMPTS:
self.redis.setex(f"security_question_locked:{user_id}", self.LOCKOUT_DURATION, "1")
return False, "Too many attempts. Account locked."
user = User.query.get(user_id)
stored_normalized = user.security_answer.lower().strip()
provided_normalized = provided_answer.lower().strip()
if stored_normalized == provided_normalized:
self.redis.delete(attempt_key)
return True, None
return False, f"Incorrect answer. {self.MAX_ATTEMPTS - attempts} attempts remaining"
def require_multiple_questions(self, user_id):
"""Require multiple questions to be answered"""
pass