DDoS #

# Contoh Slowloris — satu mesin bisa membuat server tidak responsif
# Attacker mengirim HTTP header perlahan-lahan, tidak pernah menyelesaikan request

# Simulasi konseptual (jangan digunakan untuk serangan):
# import socket
# sockets = []
# for i in range(200):
#     s = socket.socket()
#     s.connect((target, 80))
#     s.send(b"GET / HTTP/1.1\r\n")
#     s.send(b"Host: target.com\r\n")
#     sockets.append(s)
#
# while True:
#     for s in sockets:
#         s.send(b"X-a: b\r\n")  # kirim header satu per satu, sangat lambat
#     time.sleep(15)

# Server menyimpan koneksi ini tetap terbuka karena header belum selesai
# 200 koneksi bisa mematikan server Apache dengan MaxClients kecil

# Mitigasi Slowloris:
# → Nginx lebih tahan dari Apache (event-driven, bukan thread-per-connection)
# → Request timeout yang ketat
# → Limit koneksi per IP

# Implementasi rate limiting berlapis dengan Redis

import redis
import time
from functools import wraps

redis_client = redis.Redis(host='redis', port=6379, decode_responses=True)

class RateLimiter:
    """
    Rate limiter menggunakan sliding window algorithm.
    Lebih akurat dari fixed window — tidak ada spike di batas window.
    """

    def __init__(self, redis_client, max_requests: int, window_seconds: int):
        self.redis = redis_client
        self.max_requests = max_requests
        self.window = window_seconds

    def is_allowed(self, identifier: str) -> tuple[bool, dict]:
        """
        Returns: (is_allowed, rate_limit_headers)
        """
        now = time.time()
        key = f"rate:{identifier}"

        pipe = self.redis.pipeline()
        # Hapus entries yang sudah di luar window
        pipe.zremrangebyscore(key, 0, now - self.window)
        # Hitung request dalam window saat ini
        pipe.zcard(key)
        # Tambahkan request saat ini
        pipe.zadd(key, {str(now): now})
        # Set expiry
        pipe.expire(key, self.window)

        _, current_count, _, _ = pipe.execute()

        is_allowed = current_count < self.max_requests
        remaining = max(0, self.max_requests - current_count - 1)
        reset_at = int(now) + self.window

        headers = {
            'X-RateLimit-Limit': str(self.max_requests),
            'X-RateLimit-Remaining': str(remaining),
            'X-RateLimit-Reset': str(reset_at),
        }

        if not is_allowed:
            headers['Retry-After'] = str(self.window)

        return is_allowed, headers

# Beberapa limiter dengan threshold berbeda untuk endpoint berbeda
# Global: semua endpoint
global_limiter = RateLimiter(redis_client, max_requests=1000, window_seconds=60)

# API endpoint sensitif: lebih ketat
api_limiter = RateLimiter(redis_client, max_requests=100, window_seconds=60)

# Auth endpoint: sangat ketat untuk mencegah brute force
auth_limiter = RateLimiter(redis_client, max_requests=10, window_seconds=300)

def rate_limit(limiter: RateLimiter, identifier_func=None):
    """Decorator untuk rate limiting."""
    def decorator(f):
        @wraps(f)
        def decorated(*args, **kwargs):
            # Identifier: bisa per IP, per user, atau kombinasi
            if identifier_func:
                identifier = identifier_func()
            else:
                identifier = request.remote_addr

            allowed, headers = limiter.is_allowed(identifier)

            if not allowed:
                response = jsonify({
                    'error': 'Rate limit exceeded',
                    'message': 'Terlalu banyak request. Coba lagi nanti.'
                })
                response.status_code = 429
                for k, v in headers.items():
                    response.headers[k] = v
                return response

            response = f(*args, **kwargs)
            # Tambahkan rate limit info ke response header
            if hasattr(response, 'headers'):
                for k, v in headers.items():
                    response.headers[k] = v
            return response
        return decorated
    return decorator

# Penggunaan:
@app.route('/api/data')
@rate_limit(api_limiter)
def get_data():
    return jsonify(data)

@app.route('/login', methods=['POST'])
@rate_limit(auth_limiter)
def login():
    return handle_login()

# Rate limit per user (authenticated):
def get_user_identifier():
    if current_user.is_authenticated:
        return f"user:{current_user.id}"
    return f"ip:{request.remote_addr}"

@app.route('/api/expensive-operation')
@login_required
@rate_limit(api_limiter, identifier_func=get_user_identifier)
def expensive_operation():
    pass

from enum import Enum
import threading
import time

class CircuitState(Enum):
    CLOSED = "closed"      # Normal — request diizinkan
    OPEN = "open"          # Gagal — request langsung ditolak
    HALF_OPEN = "half_open"  # Testing recovery — beberapa request diizinkan

class CircuitBreaker:
    """
    Circuit breaker untuk melindungi external calls dari cascade failure.
    """

    def __init__(
        self,
        failure_threshold: int = 5,    # gagal N kali → OPEN
        recovery_timeout: int = 60,     # tunggu N detik sebelum HALF_OPEN
        success_threshold: int = 2,     # berhasil N kali di HALF_OPEN → CLOSED
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.success_threshold = success_threshold

        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time = None
        self._lock = threading.Lock()

    def call(self, func, *args, **kwargs):
        with self._lock:
            if self.state == CircuitState.OPEN:
                # Cek apakah sudah waktunya recovery
                if (time.time() - self.last_failure_time) > self.recovery_timeout:
                    self.state = CircuitState.HALF_OPEN
                    self.success_count = 0
                else:
                    raise CircuitOpenError(
                        "Circuit breaker open — service tidak tersedia"
                    )

        try:
            result = func(*args, **kwargs)

            with self._lock:
                if self.state == CircuitState.HALF_OPEN:
                    self.success_count += 1
                    if self.success_count >= self.success_threshold:
                        self.state = CircuitState.CLOSED
                        self.failure_count = 0

            return result

        except Exception as e:
            with self._lock:
                self.failure_count += 1
                self.last_failure_time = time.time()

                if self.failure_count >= self.failure_threshold:
                    self.state = CircuitState.OPEN
                    logger.warning(
                        f"Circuit breaker OPEN: too many failures",
                        extra={'failure_count': self.failure_count}
                    )
            raise

# Penggunaan:
db_circuit = CircuitBreaker(failure_threshold=5, recovery_timeout=30)
payment_circuit = CircuitBreaker(failure_threshold=3, recovery_timeout=60)

def get_user_from_db(user_id):
    return db_circuit.call(lambda: User.query.get(user_id))

def process_payment(amount, card):
    return payment_circuit.call(payment_gateway.charge, amount, card)

from functools import wraps
import redis
import json
import hashlib

redis_client = redis.Redis(host='redis', port=6379)

def cache_response(ttl: int = 60, vary_on: list = None):
    """
    Cache response endpoint.
    vary_on: list parameter yang mempengaruhi cache key
    """
    def decorator(f):
        @wraps(f)
        def decorated(*args, **kwargs):
            # Build cache key dari endpoint + parameter
            key_parts = [request.path]

            if vary_on:
                for param in vary_on:
                    value = request.args.get(param, '')
                    key_parts.append(f"{param}={value}")
            else:
                # Default: vary on semua query params
                sorted_params = sorted(request.args.items())
                key_parts.extend(f"{k}={v}" for k, v in sorted_params)

            cache_key = "cache:" + hashlib.md5(
                "|".join(key_parts).encode()
            ).hexdigest()

            # Cek cache
            cached = redis_client.get(cache_key)
            if cached:
                response_data = json.loads(cached)
                response = jsonify(response_data)
                response.headers['X-Cache'] = 'HIT'
                return response

            # Cache miss — jalankan fungsi
            response = f(*args, **kwargs)
            response.headers['X-Cache'] = 'MISS'

            # Simpan ke cache
            if response.status_code == 200:
                redis_client.setex(
                    cache_key,
                    ttl,
                    json.dumps(response.get_json())
                )

            return response
        return decorated
    return decorator

# Endpoint publik yang sering diakses — cache agresif
@app.route('/products')
@cache_response(ttl=300, vary_on=['category', 'page'])  # cache 5 menit
def list_products():
    products = Product.query.filter_by(active=True).all()
    return jsonify([p.to_dict() for p in products])

# Endpoint statis — cache sangat lama
@app.route('/api/config')
@cache_response(ttl=3600)  # cache 1 jam
def get_config():
    return jsonify(app_config)

import statistics
from collections import defaultdict, deque
from datetime import datetime, timedelta

class TrafficMonitor:
    """Monitor traffic dan deteksi anomali."""

    def __init__(self, window_size: int = 60, threshold_multiplier: float = 3.0):
        self.window_size = window_size  # detik
        self.threshold_multiplier = threshold_multiplier
        self.request_counts = deque()  # (timestamp, count)
        self.baseline_rps = None  # requests per second baseline

    def record_requests(self, count: int = 1):
        now = time.time()
        self.request_counts.append((now, count))
        # Hapus data di luar window
        while self.request_counts and \
              now - self.request_counts[0][0] > self.window_size:
            self.request_counts.popleft()

    def get_current_rps(self) -> float:
        if not self.request_counts:
            return 0
        total = sum(count for _, count in self.request_counts)
        return total / self.window_size

    def is_anomalous(self) -> bool:
        current_rps = self.get_current_rps()
        if self.baseline_rps is None:
            return False
        # Anomali jika traffic melebihi N kali baseline
        return current_rps > self.baseline_rps * self.threshold_multiplier

    def update_baseline(self, rps: float):
        """Update baseline dari periode traffic normal."""
        if self.baseline_rps is None:
            self.baseline_rps = rps
        else:
            # Exponential moving average
            self.baseline_rps = 0.9 * self.baseline_rps + 0.1 * rps

monitor = TrafficMonitor(threshold_multiplier=5.0)

# Deteksi berdasarkan distribusi IP
class IPAnomalyDetector:
    def __init__(self, window_seconds: int = 60):
        self.window = window_seconds
        self.ip_counts = defaultdict(lambda: deque())

    def record_request(self, ip: str):
        now = time.time()
        self.ip_counts[ip].append(now)
        # Hapus data lama
        while self.ip_counts[ip] and \
              now - self.ip_counts[ip][0] > self.window:
            self.ip_counts[ip].popleft()

    def get_suspicious_ips(self, threshold: int = 500) -> list[str]:
        """Return IP dengan lebih dari threshold request dalam window."""
        suspicious = []
        for ip, timestamps in self.ip_counts.items():
            if len(timestamps) > threshold:
                suspicious.append({
                    'ip': ip,
                    'count': len(timestamps),
                    'rate': len(timestamps) / self.window
                })
        return sorted(suspicious, key=lambda x: x['count'], reverse=True)

# Middleware untuk monitoring
ip_detector = IPAnomalyDetector()

@app.before_request
def monitor_traffic():
    ip = request.remote_addr
    monitor.record_requests()
    ip_detector.record_request(ip)

    # Cek anomali setiap N request untuk menghindari overhead
    if monitor.is_anomalous():
        security_logger.warning(
            "Traffic anomaly detected",
            extra={
                'current_rps': monitor.get_current_rps(),
                'baseline_rps': monitor.baseline_rps,
                'timestamp': datetime.utcnow().isoformat()
            }
        )

    # Block IP yang sangat agresif (bisa juga di nginx level)
    suspicious = ip_detector.get_suspicious_ips(threshold=200)
    if any(s['ip'] == ip for s in suspicious):
        return jsonify({'error': 'Too many requests from your IP'}), 429

# Nginx — konfigurasi untuk DDoS resistance

# Rate limiting di level nginx
limit_req_zone $binary_remote_addr zone=api:10m rate=100r/m;
limit_req_zone $binary_remote_addr zone=login:10m rate=5r/m;
limit_conn_zone $binary_remote_addr zone=conn_per_ip:10m;

server {
    listen 443 ssl;

    # Batasi koneksi per IP
    limit_conn conn_per_ip 20;

    # Timeout yang ketat — mencegah Slowloris
    client_body_timeout 10s;
    client_header_timeout 10s;
    keepalive_timeout 30s;
    send_timeout 10s;

    # Batasi ukuran request body
    client_max_body_size 10m;

    # Return 444 (close connection tanpa response) untuk user agent mencurigakan
    if ($http_user_agent ~* (bot|crawler|spider|scraper|scan)) {
        return 444;
    }

    location /api/ {
        limit_req zone=api burst=20 nodelay;
        limit_req_status 429;

        # Proxy ke aplikasi
        proxy_pass http://app:8000;
        proxy_read_timeout 30s;
        proxy_connect_timeout 10s;
    }

    location /login {
        limit_req zone=login burst=3 nodelay;
        proxy_pass http://app:8000;
    }

    # Cache static assets di CDN/browser
    location /static/ {
        expires 7d;
        add_header Cache-Control "public, immutable";
    }
}

# CDN dan DDoS protection (Cloudflare sebagai contoh)
# Konfigurasi via Terraform atau dashboard

# Prinsip konfigurasi:
# 1. Always-on DDoS protection: traffic selalu melalui CDN/scrubbing center
# 2. Challenge suspicious traffic: CAPTCHA atau JS challenge untuk IP mencurigakan
# 3. IP reputation blocking: block IP yang diketahui berbahaya
# 4. Rate limiting di edge: sebelum traffic sampai ke origin server
# 5. Geo-blocking: jika traffic dari country tertentu tidak relevan secara bisnis

# Konfigurasi rate limit di CDN (konseptual):
# - /api/*: 1000 request/menit per IP
# - /login: 10 request/menit per IP
# - /register: 5 request/menit per IP
# - Static assets: tidak di-limit (dilayani dari cache)

# Endpoint yang rentan resource exhaustion:
@app.route('/search')
def search():
    # Query yang kompleks tanpa batas
    query = request.args.get('q', '')
    results = Product.query.filter(
        Product.description.contains(query)
    ).all()  # full table scan pada tabel besar
    return jsonify([p.to_dict() for p in results])

# Jika attacker kirim 50 request bersamaan ke endpoint ini → database overload

# BENAR: endpoint yang resource-aware
@app.route('/search')
@rate_limit(api_limiter)  # rate limit
def search_safe():
    query = request.args.get('q', '').strip()

    # Validasi minimum panjang query
    if len(query) < 3:
        return jsonify({'error': 'Query minimal 3 karakter'}), 400

    # Batasi panjang query
    if len(query) > 100:
        return jsonify({'error': 'Query terlalu panjang'}), 400

    # Gunakan full-text index, bukan LIKE %...%
    # Limit hasil
    results = Product.query\
        .filter(Product.fts_vector.match(query))\
        .limit(50)\
        .all()

    return jsonify([p.to_dict() for p in results])

# Operasi yang expensive harus di-queue, bukan diproses synchronous
@app.route('/reports/generate', methods=['POST'])
@login_required
@rate_limit(RateLimiter(redis_client, max_requests=3, window_seconds=3600))  # 3x/jam
def generate_report():
    # Tidak proses langsung — masukkan ke queue
    job = report_queue.enqueue(
        generate_report_task,
        current_user.id,
        request.json,
        job_timeout=300  # maksimum 5 menit
    )
    return jsonify({
        'job_id': job.id,
        'status': 'queued',
        'check_url': f'/reports/status/{job.id}'
    }), 202  # 202 Accepted

# ✗ Anti-pattern 1: tidak ada rate limiting sama sekali
# Setiap endpoint bisa diakses tanpa batas
# Satu client bisa membuat ribuan request per detik

# ✗ Anti-pattern 2: rate limiting yang terlalu mudah dibypass
# Limit hanya berdasarkan IP — mudah dibypass dengan banyak IP
# atau dengan menggunakan proxy/VPN
# ✓ Solusi: kombinasi IP + user ID + fingerprinting

# ✗ Anti-pattern 3: synchronous processing untuk operasi mahal
@app.route('/export/csv')
def export_csv():
    # Export 1 juta baris secara synchronous
    data = Order.query.all()  # OOM risk
    return generate_csv(data)  # timeout risk
# ✓ Solusi: queue ke background worker, return job ID

# ✗ Anti-pattern 4: tidak ada timeout untuk downstream dependency
def get_external_data():
    response = requests.get('https://external-api.com/data')
    # Tidak ada timeout — jika external API lambat, thread hanging
    return response.json()
# ✓ Solusi: selalu set timeout
response = requests.get(url, timeout=(3.05, 10))  # connect timeout, read timeout

# ✗ Anti-pattern 5: cache yang bisa di-abuse untuk cache poisoning
@app.route('/user/<user_id>/profile')
@cache_response(ttl=3600)  # cache per user_id
def get_profile(user_id):
    # Jika tidak validasi ownership, attacker bisa cache
    # response dari user lain dan serve ke user lain
    return User.query.get(user_id).to_dict()
# ✓ Solusi: cache key harus include authenticated user ID untuk data private

# ✗ Anti-pattern 6: tidak ada monitoring traffic
# Spike DDoS baru ketahuan setelah sistem down
# ✓ Solusi: alerting real-time untuk RPS anomaly