Session Hijacking

Session Hijacking tek bir saldırı değil, bir saldırı sınıfıdır (attack class) . Kullanıcının oturumunun ele geçirilmesiyle sonuçlanan tüm saldırı vektörlerini kapsar:

XSS (client-side theft)

MITM (network interception)

Malware (endpoint compromise)

Session fixation

Token replay

API token leakage

Modern Sistemlerde Oturum Yönetimi
Eski Model (Stateful):
text
Cookie = session ID
Server stateful session (memory/database)
Modern Model (Stateless):
text
JWT / opaque tokens
Stateless authentication
Refresh token rotation
Kritik Gerçek: JWT kullanmak session hijacking riskini azaltmaz, sadece şeklini değiştirir. Token'lar hala çalınabilir, replay edilebilir ve kötüye kullanılabilir.

Session Binding (Modern Yaklaşım)
Eski/Yanlış Yaklaşım:
python
# IP + User-Agent binding (production'da sorunlu!)
session['ip_address'] = request.remote_addr
session['user_agent'] = request.user_agent.string
Neden Sorunlu?

Mobile IP değişir (sık sık)

VPN / CDN / NAT kullanımı

User-Agent spoof edilebilir

Aynı kullanıcı farklı cihazlardan erişebilir

Modern Yaklaşım (Weak vs Strong Binding):
Weak Binding (Eski):

IP adresi

User-Agent

Strong Binding (Modern):

Device fingerprint

Behavioral signals

Risk scoring

Session version

python
import hashlib
import json

def generate_device_fingerprint(request):
# Çoklu sinyal birleştirme
signals = {
'user_agent': request.user_agent.string,
'accept_language': request.accept_languages,
'screen_resolution': request.headers.get('X-Screen-Resolution'),
'timezone': request.headers.get('X-Timezone'),
'platform': request.headers.get('X-Platform')
}

# Fingerprint oluştur
fingerprint_data = json.dumps(signals, sort_keys=True)
return hashlib.sha256(fingerprint_data.encode()).hexdigest()

def validate_session_binding(session, request):
current_fingerprint = generate_device_fingerprint(request)
stored_fingerprint = session.get('device_fingerprint')

# Device fingerprint kontrolü (soft binding)
if stored_fingerprint and stored_fingerprint != current_fingerprint:
# Risk skorunu artır, hemen kill yapma
risk_score = calculate_risk_score(session, request)
if risk_score > 0.8:
kill_session(session)
return False

# Session version kontrolü (password change, etc.)
if session.get('session_version') != get_current_session_version(session['user_id']):
kill_session(session)
return False

return True
JWT Güvenliği (Modern Production Standartları)
Eksiksiz JWT Payload:
json
{
"sub": "user_id",
"jti": "unique-token-id-123",
"device_id": "device-fingerprint-hash",
"session_version": 3,
"iat": 1616235422,
"exp": 1616236322,
"scope": "read:profile write:settings"
}
Refresh Token Rotation (Kritik!):
python
import jwt
import uuid
import redis
from datetime import datetime, timedelta

class TokenManager:
def __init__(self):
self.redis = redis.Redis(host='localhost', port=6379, db=0)
self.access_secret = os.environ.get('JWT_ACCESS_SECRET')
self.refresh_secret = os.environ.get('JWT_REFRESH_SECRET')

def create_token_pair(self, user_id, device_id):
# Access token (kısa ömürlü - 15 dakika)
access_payload = {
'sub': str(user_id),
'jti': str(uuid.uuid4()),
'device_id': device_id,
'session_version': self.get_session_version(user_id),
'iat': datetime.utcnow(),
'exp': datetime.utcnow() + timedelta(minutes=15),
'type': 'access'
}
access_token = jwt.encode(access_payload, self.access_secret, algorithm='HS256')

# Refresh token (uzun ömürlü - 7 gün)
refresh_payload = {
'sub': str(user_id),
'jti': str(uuid.uuid4()),
'device_id': device_id,
'iat': datetime.utcnow(),
'exp': datetime.utcnow() + timedelta(days=7),
'type': 'refresh',
'access_jti': access_payload['jti'] # Access token ile bağlantı
}
refresh_token = jwt.encode(refresh_payload, self.refresh_secret, algorithm='HS256')

# Refresh token'ı Redis'te sakla
key = f"refresh:{refresh_payload['jti']}"
self.redis.setex(
key,
7 * 24 * 3600, # 7 gün
json.dumps({
'user_id': user_id,
'device_id': device_id,
'access_jti': access_payload['jti']
})
)

return access_token, refresh_token

def refresh_access_token(self, refresh_token):
try:
# Refresh token'ı doğrula
payload = jwt.decode(refresh_token, self.refresh_secret, algorithms=['HS256'])

if payload.get('type') != 'refresh':
raise Exception('Geçersiz token tipi')

# Refresh token Redis'te mi kontrol et
key = f"refresh:{payload['jti']}"
stored_data = self.redis.get(key)
if not stored_data:
raise Exception('Refresh token geçersiz veya kullanılmış')

# Refresh token rotation: eski token'ı iptal et
self.redis.delete(key)

# Yeni token çifti oluştur
user_id = payload['sub']
device_id = payload['device_id']
new_access, new_refresh = self.create_token_pair(user_id, device_id)

return new_access, new_refresh

except jwt.ExpiredSignatureError:
raise Exception('Refresh token süresi doldu')
except jwt.InvalidTokenError:
raise Exception('Geçersiz refresh token')

def revoke_all_tokens(self, user_id):
# Kullanıcının tüm token'larını iptal et
pattern = f"refresh:*"
for key in self.redis.scan_iter(pattern):
data = json.loads(self.redis.get(key))
if data['user_id'] == user_id:
self.redis.delete(key)

# Session version'ı artır
self.increment_session_version(user_id)
Token Reuse Detection:
python
def detect_token_reuse(token_jti, user_id):
"""
Aynı token'ın birden fazla kez kullanılmasını tespit et
"""
redis_client = redis.Redis()
key = f"token_used:{token_jti}"

# Token zaten kullanılmış mı?
if redis_client.exists(key):
# Token reuse tespit edildi - tüm oturumları iptal et
revoke_all_user_sessions(user_id)
raise Exception('Token reuse detected - session revoked')

# Token'ı kullanıldı olarak işaretle
redis_client.setex(key, 3600, 'used') # 1 saat boyunca cache'te tut
return True
Session Replay Attacks (Modern Tehdit)
Saldırı Senaryosu:
Saldırgan cookie çalmaz, sadece bir request yakalar ve aynısını tekrar oynatır.

text
Authorization: Bearer abc123
Koruma:
python
class RequestBindingManager:
def __init__(self):
self.redis = redis.Redis()

def create_request_binding(self, request):
# Her istek için benzersiz nonce
nonce = str(uuid.uuid4())
jti = request.jwt_payload.get('jti')

# Nonce'ı token ve istek detaylarıyla bağla
binding_data = {
'jti': jti,
'path': request.path,
'method': request.method,
'timestamp': time.time()
}

key = f"request_binding:{nonce}"
self.redis.setex(key, 300, json.dumps(binding_data)) # 5 dakika
return nonce

def validate_request_binding(self, nonce, request):
key = f"request_binding:{nonce}"
data = self.redis.get(key)

if not data:
return False

binding = json.loads(data)

# Aynı request tekrar mı?
if binding['jti'] != request.jwt_payload.get('jti'):
return False

# Path ve method eşleşiyor mu?
if binding['path'] != request.path or binding['method'] != request.method:
return False

# Nonce'ı tüket
self.redis.delete(key)
return True
Modern Defense in Depth Stack:
yaml
Layer 1: WAF (Cloudflare / AWS WAF)
- Bot detection
- Malicious request filtering
- Rate limiting

Layer 2: Bot Detection
- CAPTCHA challenges
- Browser fingerprinting
- Behavioral analysis

Layer 3: Rate Limiting
- IP-based limiting
- User-based limiting
- API endpoint limiting

Layer 4: Session Security
- Secure cookies (HttpOnly, Secure, SameSite)
- Short-lived tokens (15 min)
- Refresh token rotation
- Token reuse detection

Layer 5: Behavioral Anomaly Detection
- Impossible travel detection
- Unusual activity patterns
- Device fingerprint changes
- Session concurrency limits
Impossible Travel Detection:
python
from datetime import datetime, timedelta
import geopy.distance

class ImpossibleTravelDetector:
def __init__(self):
self.redis = redis.Redis()

def validate_location(self, user_id, current_ip):
# Kullanıcının önceki lokasyonunu al
key = f"user_location:{user_id}"
previous_data = self.redis.get(key)

if previous_data:
prev = json.loads(previous_data)
prev_lat = prev['latitude']
prev_lon = prev['longitude']
prev_time = datetime.fromisoformat(prev['timestamp'])

# Mevcut lokasyonu bul
current_lat, current_lon = self.get_location_from_ip(current_ip)
current_time = datetime.utcnow()

# Mesafe hesapla
distance = geopy.distance.distance(
(prev_lat, prev_lon),
(current_lat, current_lon)
).km

# Zaman farkı
time_diff = (current_time - prev_time).total_seconds() / 3600 # saat

# Maksimum hız: 900 km/saat (uçak)
max_distance = time_diff * 900

if distance > max_distance:
# Impossible travel tespit edildi
self.kill_session(user_id)
return False

# Yeni lokasyonu kaydet
self.redis.setex(
key,
3600,
json.dumps({
'latitude': current_lat,
'longitude': current_lon,
'timestamp': datetime.utcnow().isoformat()
})
)

return True

def kill_session(self, user_id):
# Tüm oturumları iptal et
redis_client = redis.Redis()
pattern = f"session:*"
for key in redis_client.scan_iter(pattern):
data = json.loads(redis_client.get(key))
if data.get('user_id') == user_id:
redis_client.delete(key)
CSRF vs Session Hijacking (Ayrım):
Attack Hedef Mekanizma
Session Hijacking Session ele geçirme Token/cookie çalma, replay
CSRF Session kullanarak işlem yaptırma Kullanıcının oturumuyla yetkisiz istek gönderme
Session Hijacking Kontrol Listesi (Production):
A. Token Güvenliği:
JWT kısa ömürlü mü? (5-15 dakika)

Refresh token rotation var mı?

Token reuse detection var mı?

jti (JWT ID) claim'i var mı?

device_id claim'i var mı?

session_version claim'i var mı?

B. Cookie Güvenliği:
Secure flag var mı? (HTTPS)

HttpOnly flag var mı?

SameSite=Strict veya Lax var mı?

__Secure- prefix'i var mı?

C. Binding ve Anomaly Detection:
Device fingerprint kullanılıyor mu?

Impossible travel detection var mı?

Session concurrency limit var mı?

Behavioral anomaly detection var mı?

D. Defense in Depth:
WAF var mı?

Bot detection var mı?

Rate limiting var mı?

Security headers (HSTS, etc.) var mı?

E. İzleme ve Log:
Token reuse loglanıyor mu?

Impossible travel loglanıyor mu?

Anomaly detection loglanıyor mu?

Session revocation loglanıyor mu?

🔒 Bu içeriği görmek için giriş yapın

 
Yanıt yazmak için giriş yapmalısınız
Forum özelliklerini kullanmak ve Level 2 üyelik satın almak için hesabınıza giriş yapın.
133,685Konular
3,288,873Mesajlar
324,401Kullanıcılar
Üst Alt