import requests
from flask import Flask, request
@app.route('/fetch_data')
def fetch_data():
# Vulnerable: No timeout, can hang indefinitely
url = 'https://api.example.com/data'
response = requests.get(url) # Missing timeout
return response.json()
@app.route('/post_data', methods=['POST'])
def post_data():
# Vulnerable: POST without timeout
api_url = 'https://webhook.example.com/notify'
data = request.get_json()
response = requests.post(api_url, json=data) # No timeout
return {'status': response.status_code}
@app.route('/multiple_requests')
def multiple_requests():
# Vulnerable: Multiple requests without timeouts
urls = [
'https://api1.example.com/status',
'https://api2.example.com/health',
'https://api3.example.com/ping'
]
results = []
for url in urls:
# Each request can hang indefinitely
response = requests.get(url)
results.append(response.status_code)
return {'statuses': results}
import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
from flask import Flask, request
import time
# Configure secure session with timeouts
def create_secure_session():
"""Create requests session with secure defaults."""
session = requests.Session()
# Configure retry strategy
retry_strategy = Retry(
total=3,
status_forcelist=[429, 500, 502, 503, 504],
method_whitelist=['HEAD', 'GET', 'OPTIONS'],
backoff_factor=1
)
# Mount adapter with retry strategy
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount('http://', adapter)
session.mount('https://', adapter)
return session
@app.route('/fetch_data')
def fetch_data():
"""Secure data fetching with timeout."""
try:
url = 'https://api.example.com/data'
# Secure: Explicit timeout (connect, read)
response = requests.get(
url,
timeout=(5, 10), # 5s connect, 10s read
verify=True
)
response.raise_for_status()
return response.json()
except requests.exceptions.Timeout:
return {'error': 'Request timeout'}, 408
except requests.exceptions.RequestException as e:
return {'error': f'Request failed: {str(e)}'}, 500
@app.route('/post_data', methods=['POST'])
def post_data():
"""Secure POST with timeout and validation."""
try:
api_url = 'https://webhook.example.com/notify'
data = request.get_json()
if not data:
return {'error': 'No data provided'}, 400
# Secure: POST with timeout
response = requests.post(
api_url,
json=data,
timeout=15, # 15 second timeout
headers={'Content-Type': 'application/json'},
verify=True
)
response.raise_for_status()
return {
'status': 'success',
'response_code': response.status_code,
'response_time': response.elapsed.total_seconds()
}
except requests.exceptions.Timeout:
return {'error': 'Webhook timeout'}, 408
except requests.exceptions.RequestException as e:
return {'error': f'Webhook failed: {str(e)}'}, 500
@app.route('/multiple_requests')
def multiple_requests():
"""Secure multiple requests with session and timeouts."""
urls = [
'https://api1.example.com/status',
'https://api2.example.com/health',
'https://api3.example.com/ping'
]
results = []
# Use session for connection pooling
with create_secure_session() as session:
for url in urls:
try:
start_time = time.time()
# Secure: Each request has timeout
response = session.get(
url,
timeout=(3, 5), # 3s connect, 5s read
verify=True
)
elapsed = time.time() - start_time
results.append({
'url': url,
'status_code': response.status_code,
'response_time': elapsed,
'success': True
})
except requests.exceptions.Timeout:
results.append({
'url': url,
'error': 'timeout',
'success': False
})
except requests.exceptions.RequestException as e:
results.append({
'url': url,
'error': str(e),
'success': False
})
return {
'results': results,
'successful': len([r for r in results if r.get('success')])
}
# Advanced: Custom timeout decorator
def with_timeout(connect_timeout=5, read_timeout=10):
"""Decorator to ensure requests have timeouts."""
def decorator(func):
def wrapper(*args, **kwargs):
# Inject timeout if not specified
if 'timeout' not in kwargs:
kwargs['timeout'] = (connect_timeout, read_timeout)
return func(*args, **kwargs)
return wrapper
return decorator
@with_timeout(connect_timeout=3, read_timeout=8)
def safe_get(url, **kwargs):
"""GET request with guaranteed timeout."""
return requests.get(url, **kwargs)
@app.route('/safe_request_example')
def safe_request_example():
"""Example using timeout decorator."""
try:
response = safe_get('https://api.example.com/data')
return {'status': response.status_code}
except requests.exceptions.Timeout:
return {'error': 'Request timeout'}, 408