backend/app/extensions.py - master - DnTyr/Password-Development

backend/app/extensions.py
77 строк2 KB
first_commit
03 июн 2026, 10:22
03 июн 2026, 10:22357cb35
import os
import logging
import time
from flask_limiter import Limiter

_storage_uri_cache = None
_flask_limiter_logger = logging.getLogger("flask-limiter")
_flask_limiter_logger.setLevel(logging.WARNING)
_limiter_redis_warn_every_sec = int(os.getenv("LIMITER_REDIS_WARN_EVERY_SEC", "60"))
_limiter_last_redis_warn_ts = 0.0

class _LimiterNoiseFilter(logging.Filter):

    def filter(self, record: logging.LogRecord) -> bool:
        global _limiter_last_redis_warn_ts

        msg = record.getMessage() if hasattr(record, "getMessage") else ""
        if "Swallowing error" in msg:
            now = time.time()
            if now - _limiter_last_redis_warn_ts < _limiter_redis_warn_every_sec:
                return False
            _limiter_last_redis_warn_ts = now

            record.levelno = logging.WARNING
            record.levelname = "WARNING"
            record.exc_info = None
            record.exc_text = None
            record.msg = "Rate limiting storage unavailable (Redis). Running fail-open (no rate limiting)."
            record.args = ()
            return True     

        return True

_flask_limiter_logger.addFilter(_LimiterNoiseFilter())
_flask_limiter_logger.propagate = True

def get_ip():
    from flask import request
    # Пытаемся взять реальный IP из заголовка, если ProxyFix не подменил remote_addr
    xff = request.headers.get('X-Forwarded-For')
    if xff:
        return xff.split(',')[0].strip()
    return request.remote_addr

def get_storage_uri():
    """Собирает URI для Redis один раз и кэширует результат"""
    global _storage_uri_cache
    if _storage_uri_cache is not None:
        return _storage_uri_cache

    pw_path = os.getenv("REDIS_PASSWORD_FILE", "/run/secrets/redis_pw")
    password = ""
    try:
        if os.path.exists(pw_path):
            with open(pw_path, "r") as f:
                password = f.read().strip()
    except Exception as e:
        logging.error(f"Could not read Redis secret: {e}")

    host = os.getenv("REDIS_HOST", "password-redis")
    port = os.getenv("REDIS_PORT", "6379")
    
    # 3. Формируем финальный URI
    if password:
        _storage_uri_cache = f"redis://:{password}@{host}:{port}/0"
    else:
        _storage_uri_cache = f"redis://{host}:{port}/0"
        
    return _storage_uri_cache

# Limiter
limiter = Limiter(
    key_func=get_ip,
    storage_uri=get_storage_uri(),
    strategy="fixed-window", 
    swallow_errors=True
)