Initial commit: Key Ajans portfolio site

security.py

@@ -0,0 +1,182 @@
+"""Güvenlik katmanı: admin config, CSRF, rate limit, sanitizasyon, başlıklar."""
+import hmac
+import json
+import os
+import re
+import secrets
+import threading
+import time
+
+from flask import abort, request, session
+
+# ---------------------------------------------------------------------------
+# Admin kimlik bilgileri PROJE DIŞINDA tutulur:
+#   ~/.config/key-ajansi/admins.json
+# (KEY_AJANSI_ADMINS ortam değişkeniyle başka bir yola taşınabilir)
+# ---------------------------------------------------------------------------
+DEFAULT_CONFIG_PATH = os.path.join(
+    os.path.expanduser("~"), ".config", "key-ajansi", "admins.json"
+)
+
+CONFIG_PATH = os.environ.get("KEY_AJANSI_ADMINS", DEFAULT_CONFIG_PATH)
+
+
+def load_server_config():
+    if not os.path.exists(CONFIG_PATH):
+        raise RuntimeError(
+            "Admin yapılandırması bulunamadı. Önce şunu çalıştırın:\n"
+            "    python3 setup_admins.py\n"
+            f"Beklenen dosya: {CONFIG_PATH}"
+        )
+    with open(CONFIG_PATH, "r", encoding="utf-8") as fh:
+        data = json.load(fh)
+    if not data.get("secret") or not isinstance(data.get("admins"), list):
+        raise RuntimeError(f"{CONFIG_PATH} dosyası hatalı; setup_admins.py ile yeniden oluşturun.")
+    return data
+
+
+# ---------------------------------------------------------------------------
+# Rate limiting (DDoS / brute-force koruması)
+# ---------------------------------------------------------------------------
+_limits = {}
+_lock = threading.Lock()
+
+
+def _prune_limits(now):
+    if len(_limits) > 10000:
+        for key in list(_limits):
+            if now - _limits[key][0] > 3600:
+                del _limits[key]
+
+
+def check_limit(scope, limit, window):
+    """scope başına IP'ye limit uygula; aşıldıysa False döner."""
+    key = f"{scope}:{request.remote_addr}"
+    now = time.time()
+    with _lock:
+        _prune_limits(now)
+        bucket = _limits.get(key)
+        if bucket is None or now - bucket[0] > window:
+            _limits[key] = [now, 1]
+            return True
+        if bucket[1] >= limit:
+            return False
+        bucket[1] += 1
+        return True
+
+
+def rate_limited(scope, limit, window):
+    def deco(fn):
+        def wrapper(*args, **kwargs):
+            if not check_limit(scope, limit, window):
+                abort(429)
+            return fn(*args, **kwargs)
+
+        wrapper.__name__ = fn.__name__
+        return wrapper
+
+    return deco
+
+
+# ---------------------------------------------------------------------------
+# CSRF koruması
+# ---------------------------------------------------------------------------
+def ensure_csrf_token():
+    token = session.get("csrf_token")
+    if not token:
+        token = secrets.token_urlsafe(32)
+        session["csrf_token"] = token
+    return token
+
+
+def validate_csrf():
+    token = request.form.get("csrf_token") or request.headers.get("X-CSRF-Token")
+    expected = session.get("csrf_token")
+    if not token or not expected:
+        return False
+    return hmac.compare_digest(str(token), str(expected))
+
+
+# ---------------------------------------------------------------------------
+# Girdi sanitizasyonu / doğrulaması (XSS, çok uzun girdiler)
+# ---------------------------------------------------------------------------
+_USERNAME_RE = re.compile(r"^[A-Za-z0-9_.-]{3,32}$")
+_EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
+_CONTROL_RE = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]")
+
+
+def clean_text(value, max_len):
+    """Kontrol karakterlerini temizle, kırp, uzunluğu sınırla."""
+    if value is None:
+        return ""
+    value = str(value)
+    value = _CONTROL_RE.sub("", value)
+    value = value.strip()
+    return value[:max_len]
+
+
+def clean_multiline(value, max_len):
+    return clean_text(value, max_len)
+
+
+def validate_username(value):
+    value = clean_text(value, 32)
+    if not _USERNAME_RE.match(value):
+        return None
+    return value
+
+
+def validate_email(value):
+    value = clean_text(value, 120)
+    if len(value) > 120 or not _EMAIL_RE.match(value):
+        return None
+    return value.lower()
+
+
+def validate_link(value):
+    value = clean_text(value, 500)
+    if not value:
+        return ""
+    if not re.match(r"^https?://[^\s]+$", value):
+        return None
+    if len(value) > 500:
+        return None
+    return value
+
+
+# ---------------------------------------------------------------------------
+# İstek öncesi / sonrası filtreler
+# ---------------------------------------------------------------------------
+def before_request():
+    if request.path.startswith("/static"):
+        return None
+    if not check_limit("global", 600, 300):
+        abort(429)
+    if request.method in ("POST", "PUT", "PATCH", "DELETE"):
+        if not validate_csrf():
+            abort(400, description="CSRF doğrulaması başarısız.")
+    return None
+
+
+CSP = (
+    "default-src 'self'; "
+    "script-src 'self'; "
+    "style-src 'self' https://fonts.googleapis.com; "
+    "font-src 'self' https://fonts.gstatic.com; "
+    "img-src 'self' data: https://picsum.photos; "
+    "connect-src 'self'; "
+    "frame-ancestors 'none'; "
+    "base-uri 'self'; form-action 'self'; object-src 'none'"
+)
+
+
+def after_request(response):
+    response.headers.setdefault("X-Content-Type-Options", "nosniff")
+    response.headers.setdefault("X-Frame-Options", "DENY")
+    response.headers.setdefault("Referrer-Policy", "strict-origin-when-cross-origin")
+    response.headers.setdefault("X-XSS-Protection", "0")
+    response.headers.setdefault("Permissions-Policy", "camera=(), geolocation=(), microphone=()")
+    response.headers.setdefault("Content-Security-Policy", CSP)
+    if request.path.startswith("/admin"):
+        response.headers["Cache-Control"] = "no-store, max-age=0"
+    return response