keyAjansPortfolyo/security.py

"""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