1478 lines
52 KiB
Python
1478 lines
52 KiB
Python
import os
|
||
import base64
|
||
import secrets
|
||
import re
|
||
import bleach
|
||
import secrets
|
||
import string
|
||
from mailer import mailer, send_mail
|
||
from migrate import run_migrations
|
||
from datetime import datetime, timedelta
|
||
from urllib.parse import quote_plus
|
||
from bleach import Linker
|
||
from flask import Flask, render_template, request, redirect, url_for, flash, jsonify, abort, session, make_response
|
||
from flask_wtf.csrf import CSRFProtect, generate_csrf
|
||
from flask_limiter import Limiter
|
||
from flask_limiter.util import get_remote_address
|
||
from redis import Redis
|
||
from cryptography.fernet import Fernet
|
||
from cryptography.hazmat.primitives import hashes
|
||
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
|
||
from dotenv import load_dotenv
|
||
import database
|
||
from security import (
|
||
record_failed_attempt, get_ip_block_info,
|
||
clear_failed_attempts, get_client_identifier,
|
||
block_ip, unblock_ip, get_blocked_ips, SECURITY_CONFIG,
|
||
check_admin_rate_limit
|
||
)
|
||
|
||
load_dotenv()
|
||
|
||
|
||
class Config:
|
||
APP_URL = os.environ.get('APP_URL', 'http://localhost:5000')
|
||
SECRET_KEY = os.environ.get('SECRET_KEY', secrets.token_hex(32))
|
||
REDIS_HOST = os.environ.get('REDIS_HOST', 'localhost')
|
||
REDIS_PORT = int(os.environ.get('REDIS_PORT', 6379))
|
||
REDIS_DB = int(os.environ.get('REDIS_DB', 0))
|
||
REDIS_PASSWORD = os.environ.get('REDIS_PASSWORD', None)
|
||
REDIS_PREFIX = 'secret:'
|
||
SECRET_TTL_SECONDS = 24 * 60 * 60
|
||
MAX_SECRET_LENGTH = 100000
|
||
SECRET_ID_LENGTH = 64
|
||
ENCRYPTION_PASSWORD = os.environ.get('ENCRYPTION_PASSWORD', secrets.token_hex(32))
|
||
SALT = os.environ.get('SALT', secrets.token_hex(16)).encode()
|
||
ENV = os.environ.get('FLASK_ENV', 'development')
|
||
MAX_JSON_SIZE = 1024 * 1024 # 1MB
|
||
|
||
DB_HOST = os.environ.get('DB_HOST', 'localhost')
|
||
DB_USER = os.environ.get('DB_USER', 'secretuser')
|
||
DB_PASSWORD = os.environ.get('DB_PASSWORD', '')
|
||
DB_NAME = os.environ.get('DB_NAME', 'secrettext')
|
||
DB_PORT = int(os.environ.get('DB_PORT', 3306))
|
||
|
||
DB_PASSWORD_ENCODED = quote_plus(DB_PASSWORD)
|
||
|
||
SQLALCHEMY_DATABASE_URI = f"mysql+pymysql://{DB_USER}:{DB_PASSWORD_ENCODED}@{DB_HOST}:{DB_PORT}/{DB_NAME}?charset=utf8mb4"
|
||
SQLALCHEMY_TRACK_MODIFICATIONS = False
|
||
SQLALCHEMY_ENGINE_OPTIONS = {
|
||
'pool_size': 10,
|
||
'pool_recycle': 3600,
|
||
'pool_pre_ping': True,
|
||
}
|
||
|
||
ADMIN_ALLOWED_IPS = os.environ.get('ADMIN_ALLOWED_IPS', '127.0.0.1,::1')
|
||
ADMIN_ALLOW_ALL = os.environ.get('ADMIN_ALLOW_ALL', 'False').lower() == 'true'
|
||
|
||
MAX_CONTENT_LENGTH = 10 * 1024 * 1024
|
||
SESSION_COOKIE_SECURE = True
|
||
SESSION_COOKIE_HTTPONLY = True
|
||
SESSION_COOKIE_SAMESITE = 'Lax'
|
||
PERMANENT_SESSION_LIFETIME = 3600
|
||
REMEMBER_COOKIE_SECURE = True
|
||
REMEMBER_COOKIE_HTTPONLY = True
|
||
|
||
WTF_CSRF_ENABLED = True
|
||
WTF_CSRF_SECRET_KEY = os.environ.get('WTF_CSRF_SECRET_KEY', secrets.token_hex(32))
|
||
WTF_CSRF_SSL_STRICT = True
|
||
WTF_CSRF_TIME_LIMIT = 3600
|
||
|
||
DEBUG = os.environ.get('DEBUG', 'False').lower() == 'true'
|
||
|
||
|
||
app = Flask(__name__)
|
||
app.config.from_object(Config)
|
||
|
||
ALLOWED_TAGS = ['p', 'br', 'strong', 'em', 'u', 'h1', 'h2', 'h3', 'ul', 'ol', 'li']
|
||
ALLOWED_ATTRIBUTES = {}
|
||
|
||
|
||
def create_admin():
|
||
with app.app_context():
|
||
users = database.get_all_users()
|
||
if not users:
|
||
default_username = os.environ.get('ADMIN_USERNAME', 'admin')
|
||
default_password = os.environ.get('ADMIN_PASSWORD', '')
|
||
|
||
# Очищаем пароль от пробелов и кавычек
|
||
default_password = default_password.strip()
|
||
default_password = default_password.strip('"').strip("'")
|
||
|
||
if not default_password or len(default_password) < 6:
|
||
default_password = secrets.token_hex(8)
|
||
print(f"Сгенерирован пароль: {default_password}")
|
||
|
||
success = database.create_user(default_username, default_password, role='admin')
|
||
|
||
if success:
|
||
try:
|
||
for key in redis_client.scan_iter("session:*"):
|
||
redis_client.delete(key)
|
||
except:
|
||
pass
|
||
print(f"\n{'=' * 60}")
|
||
print(f" СОЗДАН АДМИНИСТРАТОР:")
|
||
print(f" Логин: {default_username}")
|
||
print(f" Пароль: {default_password}")
|
||
print(f" Роль: admin")
|
||
print(f" Адрес: http://localhost:5000/login")
|
||
print(f" ВАЖНО: Очистите куки браузера перед входом!")
|
||
print(f"{'=' * 60}\n")
|
||
else:
|
||
print(f"Ошибка при создании администратора")
|
||
else:
|
||
print(f"\n{'=' * 60}")
|
||
print(f" Пользователи уже существуют")
|
||
print(f" Адрес: http://localhost:5000/login")
|
||
print(f"{'=' * 60}\n")
|
||
|
||
|
||
def get_base_url() -> str:
|
||
return app.config.get(
|
||
'APP_URL') or f"{app.config.get('APP_SCHEME', 'http')}://{app.config.get('APP_HOST', 'localhost')}:{app.config.get('APP_PORT', 5000)}"
|
||
|
||
|
||
def absolute_url(endpoint: str, **kwargs) -> str:
|
||
base_url = get_base_url()
|
||
relative_url = url_for(endpoint, **kwargs)
|
||
return f"{base_url}{relative_url}"
|
||
|
||
|
||
def custom_linkify(text):
|
||
linker = Linker(
|
||
callbacks=[
|
||
bleach.callbacks.nofollow,
|
||
],
|
||
skip_tags=None,
|
||
parse_email=True,
|
||
url_re=bleach.linkifier.URL_RE,
|
||
email_re=bleach.linkifier.EMAIL_RE
|
||
)
|
||
return linker.linkify(text)
|
||
|
||
|
||
database.db.init_app(app)
|
||
|
||
csrf = CSRFProtect(app)
|
||
limiter = Limiter(
|
||
app=app,
|
||
key_func=get_remote_address,
|
||
default_limits=["1000 per day", "200 per hour", "20 per minute"],
|
||
storage_uri="memory://"
|
||
)
|
||
|
||
|
||
def generate_secure_password(min_len: int = 12, max_len: int = 16) -> str:
|
||
lowercase = string.ascii_lowercase
|
||
uppercase = string.ascii_uppercase
|
||
digits = string.digits
|
||
special = "!@#$%^&*()-_=+"
|
||
|
||
password = [
|
||
secrets.choice(lowercase),
|
||
secrets.choice(uppercase),
|
||
secrets.choice(digits),
|
||
secrets.choice(special)
|
||
]
|
||
|
||
all_chars = lowercase + uppercase + digits + special
|
||
remaining_length = secrets.randbelow(max_len - min_len + 1) + (min_len - 4)
|
||
|
||
for _ in range(remaining_length):
|
||
password.append(secrets.choice(all_chars))
|
||
|
||
secrets.SystemRandom().shuffle(password)
|
||
return ''.join(password)
|
||
|
||
|
||
@app.template_filter('format_datetime')
|
||
def format_datetime(value, format='%Y-%m-%d %H:%M'):
|
||
if value is None:
|
||
return '-'
|
||
if isinstance(value, datetime):
|
||
return value.strftime(format)
|
||
return str(value)
|
||
|
||
|
||
@app.template_filter('format_datetime_seconds')
|
||
def format_datetime_seconds(value):
|
||
if value is None:
|
||
return '-'
|
||
if isinstance(value, datetime):
|
||
return value.strftime('%Y-%m-%d %H:%M:%S')
|
||
return str(value)
|
||
|
||
|
||
@app.context_processor
|
||
def inject_csrf_token():
|
||
return dict(csrf_token=generate_csrf())
|
||
|
||
|
||
@app.context_processor
|
||
def inject_user():
|
||
return {
|
||
'current_user': session.get('username'),
|
||
'is_admin': session.get('role') == 'admin',
|
||
'is_authenticated': session.get('authenticated', False)
|
||
}
|
||
|
||
|
||
def sanitize_input(text):
|
||
if not text:
|
||
return text
|
||
if len(text) > 100000:
|
||
text = text[:100000]
|
||
|
||
text = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]', '', text)
|
||
|
||
if not text:
|
||
return text
|
||
return bleach.clean(text, tags=ALLOWED_TAGS, attributes=ALLOWED_ATTRIBUTES, strip=True)
|
||
|
||
|
||
def sanitize_output(text):
|
||
if not text:
|
||
return text
|
||
text = text.replace('&', '&')
|
||
text = text.replace('<', '<')
|
||
text = text.replace('>', '>')
|
||
text = text.replace('"', '"')
|
||
text = text.replace("'", ''')
|
||
text = text.replace('\x00', '')
|
||
return text
|
||
|
||
|
||
def is_ip_allowed(ip: str) -> bool:
|
||
if app.config.get('ENV') == 'production' and app.config.get('ADMIN_ALLOW_ALL', False):
|
||
app.logger.warning(f"ADMIN_ALLOW_ALL enabled in production! Access denied for {ip}")
|
||
return False
|
||
|
||
if app.config.get('ADMIN_ALLOW_ALL', False):
|
||
app.logger.warning(f"ADMIN_ALLOW_ALL enabled (development only) for {ip}")
|
||
return True
|
||
|
||
allowed_ips_str = app.config.get('ADMIN_ALLOWED_IPS', '127.0.0.1,::1')
|
||
allowed_ips = [ip.strip() for ip in allowed_ips_str.split(',') if ip.strip()]
|
||
|
||
if ip in allowed_ips:
|
||
return True
|
||
|
||
for allowed_ip in allowed_ips:
|
||
if '/' in allowed_ip:
|
||
try:
|
||
import ipaddress
|
||
if ipaddress.ip_address(ip) in ipaddress.ip_network(allowed_ip, strict=False):
|
||
return True
|
||
except:
|
||
pass
|
||
return False
|
||
|
||
|
||
def login_required(f):
|
||
def wrapper(*args, **kwargs):
|
||
if not session.get('authenticated'):
|
||
flash('Для этого действия необходимо авторизоваться', 'warning')
|
||
return redirect(url_for('login'))
|
||
|
||
user_id = session.get('user_id')
|
||
if user_id:
|
||
user = database.get_user_by_id(user_id)
|
||
if not user:
|
||
session.clear()
|
||
flash('Ваша сессия устарела. Пожалуйста, войдите заново.', 'warning')
|
||
return redirect(url_for('login'))
|
||
return f(*args, **kwargs)
|
||
|
||
wrapper.__name__ = f.__name__
|
||
return wrapper
|
||
|
||
|
||
def admin_required(f):
|
||
def wrapper(*args, **kwargs):
|
||
client_ip = request.remote_addr
|
||
|
||
if not is_ip_allowed(client_ip):
|
||
database.log_security_event(
|
||
client_ip,
|
||
'admin_access_denied_ip',
|
||
f'Попытка доступа к админке с неразрешенного IP: {client_ip}'
|
||
)
|
||
return render_template('error.html', error='Доступ запрещен.'), 403
|
||
|
||
if not session.get('authenticated'):
|
||
flash('Для этого действия необходимо авторизоваться', 'warning')
|
||
return redirect(url_for('login'))
|
||
|
||
user_id = session.get('user_id')
|
||
if user_id:
|
||
user = database.get_user_by_id(user_id)
|
||
if not user:
|
||
session.clear()
|
||
flash('Ваша сессия устарела. Пожалуйста, войдите заново.', 'warning')
|
||
return redirect(url_for('login'))
|
||
|
||
if session.get('role') != 'admin':
|
||
flash('Доступ запрещен. Требуются права администратора.', 'danger')
|
||
return redirect(url_for('index'))
|
||
|
||
return f(*args, **kwargs)
|
||
|
||
wrapper.__name__ = f.__name__
|
||
return wrapper
|
||
|
||
|
||
@app.errorhandler(413)
|
||
def request_entity_too_large(error):
|
||
flash('Секрет слишком большой. Максимальный размер - 10MB', 'danger')
|
||
return redirect(url_for('index'))
|
||
|
||
|
||
@app.errorhandler(400)
|
||
def bad_request(error):
|
||
return render_template('error.html', error='Некорректный запрос'), 400
|
||
|
||
|
||
@app.errorhandler(403)
|
||
def forbidden(error):
|
||
return render_template('error.html', error='Доступ запрещён'), 403
|
||
|
||
|
||
@app.errorhandler(404)
|
||
def not_found(error):
|
||
return render_template('expired.html'), 404
|
||
|
||
|
||
@app.errorhandler(500)
|
||
def internal_error(error):
|
||
return render_template('error.html', error='Внутренняя ошибка сервера'), 500
|
||
|
||
|
||
@app.after_request
|
||
def add_security_headers(response):
|
||
response.headers['X-Frame-Options'] = 'DENY'
|
||
response.headers['X-Content-Type-Options'] = 'nosniff'
|
||
response.headers['X-XSS-Protection'] = '1; mode=block'
|
||
response.headers['Referrer-Policy'] = 'strict-origin-when-cross-origin'
|
||
response.headers['Permissions-Policy'] = 'geolocation=(), microphone=(), camera=()'
|
||
|
||
if request.is_secure:
|
||
response.headers['Strict-Transport-Security'] = 'max-age=31536000; includeSubDomains; preload'
|
||
|
||
csp = (
|
||
"default-src 'self'; "
|
||
"script-src 'self' 'unsafe-inline'; "
|
||
"style-src 'self' 'unsafe-inline'; "
|
||
"img-src 'self' data: https:; "
|
||
"font-src 'self'; "
|
||
"connect-src 'self'; "
|
||
"frame-ancestors 'none'; "
|
||
"form-action 'self'"
|
||
)
|
||
response.headers['Content-Security-Policy'] = csp
|
||
return response
|
||
|
||
|
||
@app.context_processor
|
||
def utility_processor():
|
||
return {
|
||
'now': datetime.now(),
|
||
'config': app.config,
|
||
'base_url': get_base_url,
|
||
'absolute_url': absolute_url
|
||
}
|
||
|
||
|
||
redis_client = Redis(
|
||
host=app.config['REDIS_HOST'],
|
||
port=app.config['REDIS_PORT'],
|
||
db=app.config['REDIS_DB'],
|
||
password=app.config['REDIS_PASSWORD'],
|
||
decode_responses=True
|
||
)
|
||
|
||
kdf = PBKDF2HMAC(
|
||
algorithm=hashes.SHA256(),
|
||
length=32,
|
||
salt=app.config['SALT'],
|
||
iterations=100000,
|
||
)
|
||
key = base64.urlsafe_b64encode(kdf.derive(app.config['ENCRYPTION_PASSWORD'].encode()))
|
||
cipher = Fernet(key)
|
||
|
||
|
||
def get_client_info():
|
||
return {
|
||
'ip': request.remote_addr,
|
||
'port': request.environ.get('REMOTE_PORT'),
|
||
'user_agent': sanitize_input(request.headers.get('User-Agent', 'Unknown')),
|
||
'accept_language': sanitize_input(request.headers.get('Accept-Language', '')),
|
||
'referer': sanitize_input(request.headers.get('Referer', '')),
|
||
'session_id': session.get('session_id', ''),
|
||
'timestamp': datetime.now().isoformat()
|
||
}
|
||
|
||
|
||
def generate_secret_id() -> str:
|
||
return secrets.token_urlsafe(app.config['SECRET_ID_LENGTH'])
|
||
|
||
|
||
def store_secret(plaintext: str, client_info: dict, user_id: int = None, request_note: str = None,
|
||
user_note: str = None) -> str:
|
||
secret_id = generate_secret_id()
|
||
|
||
encrypted = cipher.encrypt(plaintext.encode())
|
||
key = f"{app.config['REDIS_PREFIX']}{secret_id}"
|
||
redis_client.set(key, encrypted, ex=app.config['SECRET_TTL_SECONDS'])
|
||
|
||
if user_id:
|
||
user_exists = database.get_user_by_id(user_id)
|
||
if not user_exists:
|
||
user_id = None
|
||
|
||
clean_request_note = None
|
||
if request_note:
|
||
clean_request_note = sanitize_input(request_note)
|
||
if clean_request_note and len(clean_request_note) > 500:
|
||
clean_request_note = clean_request_note[:500]
|
||
|
||
clean_user_note = None
|
||
if user_note:
|
||
clean_user_note = sanitize_input(user_note)
|
||
if clean_user_note and len(clean_user_note) > 500:
|
||
clean_user_note = clean_user_note[:500]
|
||
|
||
database.log_creation(secret_id, client_info, user_id, clean_request_note, clean_user_note)
|
||
return secret_id
|
||
|
||
|
||
def get_and_delete_secret(secret_id: str, client_info: dict) -> str | None:
|
||
key = f"{app.config['REDIS_PREFIX']}{secret_id}"
|
||
encrypted = redis_client.get(key)
|
||
if encrypted is None:
|
||
return None
|
||
redis_client.delete(key)
|
||
try:
|
||
decrypted = cipher.decrypt(encrypted.encode() if isinstance(encrypted, str) else encrypted)
|
||
plaintext = decrypted.decode()
|
||
safe_text = sanitize_output(plaintext)
|
||
database.log_view(secret_id, client_info)
|
||
return safe_text
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def get_secret_info(secret_id: str) -> dict | None:
|
||
key = f"{app.config['REDIS_PREFIX']}{secret_id}"
|
||
ttl = redis_client.ttl(key)
|
||
if ttl <= 0:
|
||
return None
|
||
return {'exists': True, 'ttl_seconds': ttl, 'ttl_hours': round(ttl / 3600, 1)}
|
||
|
||
|
||
@app.route('/')
|
||
def index():
|
||
if session.get('authenticated'):
|
||
return render_template('index.html')
|
||
return redirect(url_for('login'))
|
||
|
||
|
||
def check_rate_limit(ip: str, key: str, limit: int = 10, window: int = 60) -> bool:
|
||
try:
|
||
redis_key = f"rate_limit:{key}:{ip}"
|
||
current = redis_client.incr(redis_key)
|
||
if current == 1:
|
||
redis_client.expire(redis_key, window)
|
||
|
||
if current > limit:
|
||
return False
|
||
|
||
return True
|
||
except Exception as e:
|
||
app.logger.warning(f"Redis unavailable for rate limiting: {key} - {e}")
|
||
return True
|
||
|
||
|
||
def get_rate_limit_remaining(ip: str, key: str, limit: int = 10, window: int = 60) -> int:
|
||
try:
|
||
redis_key = f"rate_limit:{key}:{ip}"
|
||
current = redis_client.get(redis_key)
|
||
if current:
|
||
return max(0, limit - int(current))
|
||
return limit
|
||
except Exception:
|
||
return limit
|
||
|
||
|
||
@app.route('/login', methods=['GET', 'POST'])
|
||
@limiter.limit("30 per minute")
|
||
def login():
|
||
if not check_rate_limit(request.remote_addr, 'login'):
|
||
flash('Слишком много попыток входа. Попробуйте позже.', 'danger')
|
||
return render_template('login.html'), 429
|
||
|
||
if session.get('authenticated'):
|
||
user_id = session.get('user_id')
|
||
if user_id:
|
||
user = database.get_user_by_id(user_id)
|
||
if user:
|
||
return redirect(url_for('index'))
|
||
else:
|
||
session.clear()
|
||
|
||
client_ip = request.remote_addr
|
||
|
||
if request.method == 'POST':
|
||
username = request.form.get('username', '').strip()
|
||
password = request.form.get('password', '')
|
||
|
||
if not username or not password:
|
||
flash('Введите логин и пароль', 'danger')
|
||
return render_template('login.html')
|
||
|
||
user = database.get_user(username)
|
||
if user and user.check_password(password):
|
||
if not user.is_active:
|
||
flash('Аккаунт деактивирован. Обратитесь к администратору.', 'danger')
|
||
return render_template('login.html')
|
||
|
||
session.clear()
|
||
|
||
session['authenticated'] = True
|
||
session['username'] = username
|
||
session['user_id'] = user.id
|
||
session['role'] = user.role
|
||
session.permanent = True
|
||
|
||
database.update_last_login(username)
|
||
database.log_security_event(
|
||
client_ip,
|
||
'user_login',
|
||
f'Успешный вход: {username} (role: {user.role})',
|
||
user_id=user.id
|
||
)
|
||
|
||
flash(f'Добро пожаловать, {username}!', 'success')
|
||
|
||
if user.role == 'admin':
|
||
return redirect(url_for('admin_panel'))
|
||
return redirect(url_for('index'))
|
||
|
||
database.log_security_event(
|
||
client_ip,
|
||
'user_login_failed',
|
||
f'Неудачная попытка входа: {username}'
|
||
)
|
||
flash('Неверный логин или пароль', 'danger')
|
||
return render_template('login.html')
|
||
|
||
|
||
@app.route('/logout')
|
||
def logout():
|
||
username = session.get('username', 'unknown')
|
||
user_id = session.get('user_id')
|
||
database.log_security_event(
|
||
request.remote_addr,
|
||
'user_logout',
|
||
f'Выход: {username}',
|
||
user_id=user_id
|
||
)
|
||
session.clear()
|
||
flash('Вы вышли из системы', 'info')
|
||
return redirect(url_for('index'))
|
||
|
||
|
||
@app.route('/profile', methods=['GET', 'POST'])
|
||
@login_required
|
||
def profile():
|
||
user_id = session.get('user_id')
|
||
user = database.get_user_by_id(user_id)
|
||
|
||
if not user:
|
||
session.clear()
|
||
flash('Ваша сессия устарела. Пожалуйста, войдите заново.', 'warning')
|
||
return redirect(url_for('login'))
|
||
|
||
if request.method == 'POST':
|
||
email = request.form.get('email', '').strip()
|
||
jabber = request.form.get('jabber', '').strip()
|
||
|
||
if email and not re.match(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$', email):
|
||
flash('Неверный формат email', 'danger')
|
||
return render_template('profile.html', user=user)
|
||
|
||
if jabber and not re.match(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$', jabber):
|
||
flash('Неверный формат Jabber ID', 'danger')
|
||
return render_template('profile.html', user=user)
|
||
|
||
user.email = email if email else None
|
||
user.jabber = jabber if jabber else None
|
||
user.updated_at = datetime.now()
|
||
|
||
user.notify_email = 'notify_email' in request.form
|
||
user.notify_jabber = 'notify_jabber' in request.form
|
||
|
||
database.db.session.commit()
|
||
|
||
database.log_security_event(
|
||
request.remote_addr,
|
||
'profile_updated',
|
||
f'Обновлен профиль пользователя {user.username}',
|
||
user_id=user.id
|
||
)
|
||
|
||
flash('Профиль успешно обновлен!', 'success')
|
||
return redirect(url_for('profile'))
|
||
|
||
return render_template('profile.html', user=user)
|
||
|
||
|
||
@app.route('/profile/change-password', methods=['POST'])
|
||
@login_required
|
||
def profile_change_password():
|
||
user_id = session.get('user_id')
|
||
user = database.get_user_by_id(user_id)
|
||
|
||
if not user:
|
||
session.clear()
|
||
flash('Ваша сессия устарела.', 'warning')
|
||
return redirect(url_for('login'))
|
||
|
||
old_password = request.form.get('old_password', '')
|
||
new_password = request.form.get('new_password', '')
|
||
confirm_password = request.form.get('confirm_password', '')
|
||
|
||
if not user.check_password(old_password):
|
||
flash('Неверный текущий пароль', 'danger')
|
||
return redirect(url_for('profile'))
|
||
|
||
if len(new_password) < 6:
|
||
flash('Новый пароль должен содержать минимум 6 символов', 'danger')
|
||
return redirect(url_for('profile'))
|
||
|
||
if new_password != confirm_password:
|
||
flash('Пароли не совпадают', 'danger')
|
||
return redirect(url_for('profile'))
|
||
|
||
user.set_password(new_password)
|
||
database.db.session.commit()
|
||
|
||
database.log_security_event(
|
||
request.remote_addr,
|
||
'password_change',
|
||
f'Смена пароля пользователем {user.username}',
|
||
user_id=user.id
|
||
)
|
||
|
||
flash('Пароль успешно изменен!', 'success')
|
||
return redirect(url_for('profile'))
|
||
|
||
|
||
@app.route('/create', methods=['POST'])
|
||
@login_required
|
||
@limiter.limit("30 per minute")
|
||
def create_secret():
|
||
secret_text = request.form.get('secret', '').strip()
|
||
user_note = request.form.get('user_note', '').strip()
|
||
|
||
if not secret_text:
|
||
flash('Пожалуйста, введите секрет', 'danger')
|
||
return redirect(url_for('index'))
|
||
|
||
if len(secret_text) > app.config['MAX_SECRET_LENGTH']:
|
||
flash(f'Секрет слишком длинный. Максимум {app.config["MAX_SECRET_LENGTH"]} символов', 'danger')
|
||
return redirect(url_for('index'))
|
||
|
||
if re.search(r'<script.*?>.*?</script>', secret_text, re.IGNORECASE | re.DOTALL):
|
||
flash('Обнаружено подозрительное содержимое. Секрет не сохранён.', 'danger')
|
||
return redirect(url_for('index'))
|
||
|
||
client_info = get_client_info()
|
||
user_id = session.get('user_id')
|
||
|
||
if user_id:
|
||
user_exists = database.get_user_by_id(user_id)
|
||
if not user_exists:
|
||
session.clear()
|
||
flash('Ваша сессия устарела. Пожалуйста, войдите заново.', 'warning')
|
||
return redirect(url_for('login'))
|
||
|
||
secret_id = store_secret(secret_text, client_info, user_id, request_note=user_note)
|
||
view_url = url_for('view_secret', secret_id=secret_id, _external=True)
|
||
|
||
database.log_security_event(
|
||
client_info['ip'],
|
||
'secret_created',
|
||
f'Создан секрет: {secret_id} пользователем {session.get("username")}',
|
||
secret_id=secret_id,
|
||
user_id=user_id
|
||
)
|
||
|
||
return render_template('created.html', link=view_url, secret_id=secret_id, expires_in_hours=24)
|
||
|
||
|
||
@app.route('/view/<secret_id>')
|
||
@limiter.limit("60 per minute")
|
||
def view_secret(secret_id):
|
||
if not secret_id or len(secret_id) < 10:
|
||
abort(404)
|
||
|
||
identifier, ip = get_client_identifier()
|
||
block_info = get_ip_block_info(ip)
|
||
if block_info['blocked']:
|
||
if block_info['permanent']:
|
||
return render_template('blocked.html',
|
||
reason='Ваш IP-адрес был permanently заблокирован',
|
||
permanent=True), 403
|
||
else:
|
||
return render_template('blocked.html',
|
||
reason=f'Слишком много неудачных попыток. Попробуйте через {block_info["remaining_minutes"]} минут.',
|
||
remaining=block_info['remaining_seconds'],
|
||
permanent=False), 403
|
||
|
||
key = f"{app.config['REDIS_PREFIX']}{secret_id}"
|
||
|
||
try:
|
||
encrypted = redis_client.getdel(key)
|
||
except Exception:
|
||
try:
|
||
pipeline = redis_client.pipeline()
|
||
pipeline.get(key)
|
||
pipeline.delete(key)
|
||
result = pipeline.execute()
|
||
encrypted = result[0] if result else None
|
||
except Exception as e:
|
||
app.logger.error(f"Redis error in view_secret: {e}")
|
||
return render_template('error.html', error='Техническая ошибка'), 500
|
||
|
||
if encrypted is None:
|
||
record_failed_attempt(identifier)
|
||
return render_template('expired.html'), 404
|
||
|
||
clear_failed_attempts(identifier)
|
||
client_info = get_client_info()
|
||
|
||
try:
|
||
decrypted = cipher.decrypt(encrypted.encode() if isinstance(encrypted, str) else encrypted)
|
||
plaintext = decrypted.decode()
|
||
|
||
import html
|
||
escaped_text = html.escape(plaintext)
|
||
|
||
safe_text = bleach.linkify(
|
||
escaped_text,
|
||
callbacks=[bleach.callbacks.nofollow],
|
||
parse_email=True
|
||
)
|
||
safe_text = safe_text.replace('\n', '<br>')
|
||
|
||
database.log_view(secret_id, client_info)
|
||
return render_template('view.html', secret=safe_text)
|
||
|
||
except Exception as e:
|
||
app.logger.error(f"Error processing secret {secret_id}: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return render_template('expired.html'), 404
|
||
|
||
|
||
@app.route('/my-secrets')
|
||
@login_required
|
||
def my_secrets():
|
||
user_id = session.get('user_id')
|
||
secrets = database.get_user_secrets(user_id, limit=100)
|
||
return render_template('my_secrets.html', secrets=secrets)
|
||
|
||
|
||
@app.route('/request-secret', methods=['GET'])
|
||
@login_required
|
||
def request_secret_form():
|
||
return render_template('request_secret.html')
|
||
|
||
|
||
@app.route('/request-secret/create', methods=['POST'])
|
||
@login_required
|
||
@limiter.limit("30 per minute")
|
||
def request_secret_create():
|
||
note = request.form.get('note', '').strip()
|
||
|
||
if len(note) > 500:
|
||
flash('Примечание слишком длинное (максимум 500 символов)', 'danger')
|
||
return redirect(url_for('request_secret_form'))
|
||
|
||
request_token = secrets.token_urlsafe(32)
|
||
user_id = session.get('user_id')
|
||
|
||
if user_id:
|
||
user_exists = database.get_user_by_id(user_id)
|
||
if not user_exists:
|
||
session.clear()
|
||
flash('Ваша сессия устарела. Пожалуйста, войдите заново.', 'warning')
|
||
return redirect(url_for('login'))
|
||
|
||
request_key = f"request:{request_token}"
|
||
redis_client.set(request_key, str(user_id), ex=1 * 24 * 60 * 60)
|
||
|
||
if note:
|
||
note_key = f"request_note:{request_token}"
|
||
redis_client.set(note_key, note, ex=1 * 24 * 60 * 60)
|
||
|
||
database.log_security_event(
|
||
request.remote_addr,
|
||
'secret_request_created',
|
||
f'Создан запрос секрета пользователем {session.get("username")}',
|
||
user_id=user_id
|
||
)
|
||
|
||
request_url = url_for('request_secret_submit', token=request_token, _external=True)
|
||
|
||
flash('Ссылка для запроса секрета создана! Отправьте её получателю.', 'success')
|
||
return render_template('request_secret_done.html',
|
||
link=request_url,
|
||
token=request_token,
|
||
note=note)
|
||
|
||
|
||
@app.route('/request/<token>', methods=['GET', 'POST'])
|
||
@limiter.limit("30 per minute")
|
||
def request_secret_submit(token):
|
||
request_key = f"request:{token}"
|
||
user_id = redis_client.get(request_key)
|
||
|
||
if not user_id:
|
||
flash('Ссылка для запроса секрета недействительна или истекла', 'danger')
|
||
return render_template('request_secret_expired.html')
|
||
|
||
note_key = f"request_note:{token}"
|
||
note = redis_client.get(note_key) or ''
|
||
|
||
user = database.get_user_by_id(int(user_id))
|
||
recipient_name = user.username if user else 'Неизвестный пользователь'
|
||
|
||
if request.method == 'POST':
|
||
secret_text = request.form.get('secret', '').strip()
|
||
|
||
if not secret_text:
|
||
flash('Пожалуйста, введите секрет', 'danger')
|
||
return render_template('request_secret_submit.html',
|
||
token=token,
|
||
note=note,
|
||
recipient_name=recipient_name)
|
||
|
||
if len(secret_text) > app.config['MAX_SECRET_LENGTH']:
|
||
flash(f'Секрет слишком длинный. Максимум {app.config["MAX_SECRET_LENGTH"]} символов', 'danger')
|
||
return render_template('request_secret_submit.html',
|
||
token=token,
|
||
note=note,
|
||
recipient_name=recipient_name)
|
||
|
||
if re.search(r'<script.*?>.*?</script>', secret_text, re.IGNORECASE | re.DOTALL):
|
||
flash('Обнаружено подозрительное содержимое. Секрет не сохранён.', 'danger')
|
||
return render_template('request_secret_submit.html',
|
||
token=token,
|
||
note=note,
|
||
recipient_name=recipient_name)
|
||
|
||
client_info = get_client_info()
|
||
|
||
clean_note = None
|
||
if note:
|
||
clean_note = sanitize_input(note)
|
||
if clean_note and len(clean_note) > 500:
|
||
clean_note = clean_note[:500]
|
||
|
||
secret_id = store_secret(secret_text, client_info, int(user_id), request_note=clean_note)
|
||
view_url = url_for('view_secret', secret_id=secret_id, _external=True)
|
||
|
||
redis_client.delete(request_key)
|
||
redis_client.delete(note_key)
|
||
|
||
database.log_security_event(
|
||
request.remote_addr,
|
||
'secret_submitted_from_request',
|
||
f'Секрет отправлен по запросу для пользователя {recipient_name}',
|
||
secret_id=secret_id,
|
||
user_id=int(user_id)
|
||
)
|
||
|
||
if user and user.email and mailer.is_enabled() and user.notify_email:
|
||
try:
|
||
from mailer import send_secret_received_email
|
||
send_secret_received_email(
|
||
to_email=user.email,
|
||
username=user.username,
|
||
note=clean_note
|
||
)
|
||
print(f"Уведомление о секрете отправлено на {user.email}")
|
||
except Exception as e:
|
||
print(f"Ошибка отправки уведомления: {e}")
|
||
else:
|
||
# Логируем причину, почему письмо не отправлено
|
||
reasons = []
|
||
if not user:
|
||
reasons.append("пользователь не найден")
|
||
elif not user.email:
|
||
reasons.append("email не указан")
|
||
elif not mailer.is_enabled():
|
||
reasons.append("почта отключена")
|
||
elif not user.notify_email:
|
||
reasons.append("отключены уведомления в профиле")
|
||
print(f"ℹ️ Письмо не отправлено: {', '.join(reasons)}")
|
||
|
||
flash('Секрет успешно отправлен! Получатель сможет его просмотреть.', 'success')
|
||
|
||
return render_template('request_secret_submit_done.html',
|
||
secret_id=secret_id,
|
||
recipient_name=recipient_name)
|
||
|
||
return render_template('request_secret_submit.html',
|
||
token=token,
|
||
note=note,
|
||
recipient_name=recipient_name)
|
||
|
||
|
||
@app.route('/my-requests')
|
||
@login_required
|
||
def my_requests():
|
||
user_id = session.get('user_id')
|
||
requests = []
|
||
|
||
MAX_TTL = 24 * 60 * 60 # 24 часа
|
||
|
||
cursor = 0
|
||
try:
|
||
while True:
|
||
cursor, keys = redis_client.scan(
|
||
cursor,
|
||
match="request:*",
|
||
count=100
|
||
)
|
||
|
||
for key in keys:
|
||
user_id_from_redis = redis_client.get(key)
|
||
if user_id_from_redis == str(user_id):
|
||
token = key.replace("request:", "")
|
||
ttl = redis_client.ttl(key)
|
||
|
||
if ttl > 0:
|
||
created_at = datetime.now() - timedelta(seconds=(MAX_TTL - ttl))
|
||
else:
|
||
created_at = datetime.now() - timedelta(hours=24)
|
||
|
||
note_key = f"request_note:{token}"
|
||
note = redis_client.get(note_key) or ''
|
||
|
||
import html
|
||
safe_note = html.escape(note[:500]) if note else ''
|
||
|
||
link = url_for('request_secret_submit', token=token, _external=True)
|
||
|
||
requests.append({
|
||
'token': token,
|
||
'link': link,
|
||
'note': safe_note,
|
||
'ttl_seconds': ttl,
|
||
'ttl_hours': round(ttl / 3600, 1),
|
||
'created_at': created_at
|
||
})
|
||
|
||
if cursor == 0:
|
||
break
|
||
except Exception as e:
|
||
app.logger.error(f"Redis scan error in my_requests: {e}")
|
||
flash('Ошибка при загрузке запросов', 'danger')
|
||
return render_template('my_requests.html', requests=[])
|
||
|
||
return render_template('my_requests.html', requests=requests)
|
||
|
||
|
||
@app.route('/request/<token>/delete', methods=['POST'])
|
||
@login_required
|
||
def request_secret_delete(token):
|
||
user_id = session.get('user_id')
|
||
request_key = f"request:{token}"
|
||
|
||
user_id_from_redis = redis_client.get(request_key)
|
||
if not user_id_from_redis or int(user_id_from_redis) != user_id:
|
||
flash('Запрос не найден или не принадлежит вам', 'danger')
|
||
return redirect(url_for('my_requests'))
|
||
|
||
redis_client.delete(request_key)
|
||
redis_client.delete(f"request_note:{token}")
|
||
|
||
database.log_security_event(
|
||
request.remote_addr,
|
||
'secret_request_deleted',
|
||
f'Удален запрос секрета пользователем {session.get("username")}',
|
||
user_id=user_id
|
||
)
|
||
|
||
flash('Запрос удален', 'success')
|
||
return redirect(url_for('my_requests'))
|
||
|
||
|
||
@app.route('/admin/login', methods=['GET', 'POST'])
|
||
@limiter.limit("30 per minute")
|
||
def admin_login():
|
||
client_ip = request.remote_addr
|
||
|
||
if not is_ip_allowed(client_ip):
|
||
database.log_security_event(
|
||
client_ip,
|
||
'admin_access_denied_ip',
|
||
f'Попытка доступа к странице входа с неразрешенного IP: {client_ip}'
|
||
)
|
||
return render_template('error.html', error='Доступ запрещен.'), 403
|
||
|
||
if session.get('authenticated') and session.get('role') == 'admin':
|
||
return redirect(url_for('admin_panel'))
|
||
|
||
if check_admin_rate_limit(client_ip):
|
||
flash('Слишком много попыток входа. Подождите 5 минут.', 'danger')
|
||
return render_template('admin_login.html')
|
||
|
||
if request.method == 'POST':
|
||
username = request.form.get('username', '').strip()
|
||
password = request.form.get('password', '')
|
||
|
||
if not username or not password:
|
||
flash('Введите логин и пароль', 'danger')
|
||
return render_template('admin_login.html')
|
||
|
||
user = database.get_user(username)
|
||
if user and user.check_password(password):
|
||
if not user.is_active:
|
||
flash('Аккаунт деактивирован', 'danger')
|
||
return render_template('admin_login.html')
|
||
|
||
if user.role != 'admin':
|
||
flash('Доступ запрещен. Требуются права администратора.', 'danger')
|
||
return render_template('admin_login.html')
|
||
|
||
session.clear()
|
||
session['authenticated'] = True
|
||
session['username'] = username
|
||
session['user_id'] = user.id
|
||
session['role'] = user.role
|
||
session.permanent = True
|
||
|
||
database.update_last_login(username)
|
||
database.log_security_event(
|
||
client_ip,
|
||
'admin_login',
|
||
f'Успешный вход в админку: {username}',
|
||
user_id=user.id
|
||
)
|
||
|
||
flash('Вход выполнен', 'success')
|
||
return redirect(url_for('admin_panel'))
|
||
|
||
database.log_security_event(
|
||
client_ip,
|
||
'admin_login_failed',
|
||
f'Неудачная попытка входа в админку: {username}'
|
||
)
|
||
flash('Неверный логин или пароль', 'danger')
|
||
|
||
return render_template('admin_login.html')
|
||
|
||
|
||
@app.route('/admin/logout')
|
||
def admin_logout():
|
||
username = session.get('username', 'unknown')
|
||
user_id = session.get('user_id')
|
||
database.log_security_event(
|
||
request.remote_addr,
|
||
'admin_logout',
|
||
f'Выход из админки: {username}',
|
||
user_id=user_id
|
||
)
|
||
session.clear()
|
||
flash('Вы вышли из админки', 'info')
|
||
return redirect(url_for('index'))
|
||
|
||
|
||
@app.route('/admin')
|
||
@admin_required
|
||
def admin_panel():
|
||
filter_viewed = request.args.get('filter', None)
|
||
if filter_viewed not in ['viewed', 'pending', None]:
|
||
filter_viewed = None
|
||
|
||
try:
|
||
page = request.args.get('page', 1, type=int)
|
||
per_page = 50
|
||
offset = (page - 1) * per_page
|
||
filter_viewed = request.args.get('filter', None)
|
||
|
||
logs = database.get_all_logs(limit=per_page, offset=offset, filter_viewed=filter_viewed)
|
||
stats = database.get_stats()
|
||
security_logs = database.get_security_logs(limit=20)
|
||
blocked_ips = get_blocked_ips()
|
||
users = database.get_all_users()
|
||
|
||
return render_template('admin.html',
|
||
logs=logs,
|
||
stats=stats,
|
||
security_logs=security_logs,
|
||
blocked_ips=blocked_ips,
|
||
users=users,
|
||
page=page,
|
||
per_page=per_page,
|
||
filter=filter_viewed)
|
||
except Exception as e:
|
||
print(f"Ошибка в админке: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
flash('Ошибка загрузки админки', 'danger')
|
||
return redirect(url_for('index'))
|
||
|
||
|
||
@app.route('/admin/user/create', methods=['GET', 'POST'])
|
||
@admin_required
|
||
def admin_create_user():
|
||
if request.method == 'POST':
|
||
username = request.form.get('username', '').strip()
|
||
email = request.form.get('email', '').strip()
|
||
role = request.form.get('role', 'user')
|
||
|
||
if not username or len(username) < 3:
|
||
flash('Логин должен содержать минимум 3 символа', 'danger')
|
||
return render_template('admin_user_create.html')
|
||
|
||
if not re.match(r'^[a-zA-Z0-9_.]+$', username):
|
||
flash('Логин может содержать только буквы, цифры и знак подчеркивания', 'danger')
|
||
return render_template('admin_user_create.html')
|
||
|
||
if not email:
|
||
flash('Email обязателен для заполнения', 'danger')
|
||
return render_template('admin_user_create.html')
|
||
|
||
if not re.match(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$', email):
|
||
flash('Неверный формат email', 'danger')
|
||
return render_template('admin_user_create.html')
|
||
|
||
if role not in ['user', 'admin']:
|
||
flash('Неверная роль', 'danger')
|
||
return render_template('admin_user_create.html')
|
||
|
||
existing = database.get_user(username)
|
||
if existing:
|
||
flash(f'Пользователь "{username}" уже существует', 'danger')
|
||
return render_template('admin_user_create.html')
|
||
|
||
password = generate_secure_password()
|
||
print(f"Сгенерирован пароль для {username}: {password}")
|
||
|
||
if not database.create_user(username, password, role):
|
||
flash('Ошибка при создании пользователя', 'danger')
|
||
return render_template('admin_user_create.html')
|
||
|
||
new_user = database.get_user(username)
|
||
new_user_id = new_user.id if new_user else None
|
||
print(f"✅ Пользователь {username} создан с ID: {new_user_id}")
|
||
|
||
admin_id = session.get('user_id')
|
||
admin_username = session.get('username', 'admin')
|
||
|
||
secret_text = f"""Пароль для {username}:
|
||
|
||
{password}
|
||
|
||
Логин: {username}
|
||
|
||
Отправлено на: {email}
|
||
|
||
ВАЖНО:
|
||
• Пароль был отправлен пользователю в виде одноразовой ссылки
|
||
• После прочтения ссылка станет недействительной
|
||
|
||
Создано: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
|
||
Администратор: {admin_username}
|
||
"""
|
||
|
||
client_info = {
|
||
'ip': request.remote_addr,
|
||
'user_agent': request.headers.get('User-Agent', 'Unknown'),
|
||
'session_id': session.get('session_id', '')
|
||
}
|
||
|
||
secret_id = store_secret(
|
||
plaintext=secret_text,
|
||
client_info=client_info,
|
||
user_id=admin_id, # ← ID администратора!
|
||
request_note=f"Пароль для {username} (отправлен на {email})"
|
||
)
|
||
|
||
view_url = url_for('view_secret', secret_id=secret_id, _external=True)
|
||
|
||
if mailer.is_enabled():
|
||
try:
|
||
mailer.send_user_created_email(email, username, view_url)
|
||
flash(f'Письмо с паролем отправлено на {email}', 'success')
|
||
except Exception as e:
|
||
flash(f'Ошибка отправки письма: {str(e)}', 'danger')
|
||
print(f"Пароль для {username}: {password}")
|
||
else:
|
||
flash(f'Почта отключена. Пароль для {username}: {password} (сохраните его!)', 'warning')
|
||
|
||
database.log_security_event(
|
||
request.remote_addr,
|
||
'user_created_with_email',
|
||
f'Создан пользователь: {username} (role: {role}) администратором {admin_username}. Пароль отправлен на {email}',
|
||
user_id=admin_id
|
||
)
|
||
flash(f'Пользователь "{username}" успешно создан! Пароль отправлен на email.', 'success')
|
||
return redirect(url_for('admin_panel'))
|
||
|
||
return render_template('admin_user_create.html')
|
||
|
||
|
||
@app.route('/admin/user/<int:user_id>/toggle')
|
||
@admin_required
|
||
def admin_toggle_user(user_id):
|
||
if user_id == session.get('user_id'):
|
||
flash('Нельзя деактивировать самого себя', 'danger')
|
||
return redirect(url_for('admin_panel'))
|
||
|
||
user = database.get_user_by_id(user_id)
|
||
if user:
|
||
new_status = not user.is_active
|
||
database.set_user_active(user_id, new_status)
|
||
status_text = "активирован" if new_status else "деактивирован"
|
||
database.log_security_event(
|
||
request.remote_addr,
|
||
'user_toggle',
|
||
f'Пользователь {user.username} {status_text} администратором {session.get("username")}',
|
||
user_id=session.get('user_id')
|
||
)
|
||
flash(f'Пользователь "{user.username}" {status_text}', 'success')
|
||
return redirect(url_for('admin_panel'))
|
||
|
||
|
||
@app.route('/admin/user/<int:user_id>/role', methods=['POST'])
|
||
@admin_required
|
||
def admin_change_role(user_id):
|
||
if user_id == session.get('user_id'):
|
||
flash('Нельзя изменить роль самого себя', 'danger')
|
||
return redirect(url_for('admin_panel'))
|
||
|
||
new_role = request.form.get('role', 'user')
|
||
if new_role not in ['user', 'admin']:
|
||
flash('Неверная роль', 'danger')
|
||
return redirect(url_for('admin_panel'))
|
||
|
||
user = database.get_user_by_id(user_id)
|
||
if user:
|
||
database.set_user_role(user_id, new_role)
|
||
database.log_security_event(
|
||
request.remote_addr,
|
||
'user_role_changed',
|
||
f'Роль пользователя {user.username} изменена на {new_role} администратором {session.get("username")}',
|
||
user_id=session.get('user_id')
|
||
)
|
||
flash(f'Роль пользователя "{user.username}" изменена на {new_role}', 'success')
|
||
return redirect(url_for('admin_panel'))
|
||
|
||
|
||
@app.route('/admin/user/<int:user_id>/delete')
|
||
@admin_required
|
||
def admin_delete_user(user_id):
|
||
if user_id == session.get('user_id'):
|
||
flash('Нельзя удалить самого себя', 'danger')
|
||
return redirect(url_for('admin_panel'))
|
||
|
||
user = database.get_user_by_id(user_id)
|
||
if user:
|
||
database.log_security_event(
|
||
request.remote_addr,
|
||
'user_deleted',
|
||
f'Пользователь {user.username} удалён администратором {session.get("username")}',
|
||
user_id=session.get('user_id')
|
||
)
|
||
database.delete_user(user_id)
|
||
flash(f'Пользователь "{user.username}" удалён', 'success')
|
||
return redirect(url_for('admin_panel'))
|
||
|
||
|
||
@app.route('/admin/change_password', methods=['POST'])
|
||
@admin_required
|
||
def admin_change_password():
|
||
old_password = request.form.get('old_password', '')
|
||
new_password = request.form.get('new_password', '')
|
||
confirm_password = request.form.get('confirm_password', '')
|
||
|
||
username = session.get('username')
|
||
user = database.get_user(username)
|
||
|
||
if not user or not user.check_password(old_password):
|
||
flash('Неверный текущий пароль', 'danger')
|
||
return redirect(url_for('admin_panel'))
|
||
|
||
if len(new_password) < 6:
|
||
flash('Новый пароль должен содержать минимум 6 символов', 'danger')
|
||
return redirect(url_for('admin_panel'))
|
||
|
||
if new_password != confirm_password:
|
||
flash('Пароли не совпадают', 'danger')
|
||
return redirect(url_for('admin_panel'))
|
||
|
||
user.set_password(new_password)
|
||
database.db.session.commit()
|
||
|
||
database.log_security_event(
|
||
request.remote_addr,
|
||
'password_change',
|
||
f'Смена пароля: {username}',
|
||
user_id=session.get('user_id')
|
||
)
|
||
flash('Пароль успешно изменён', 'success')
|
||
return redirect(url_for('admin_panel'))
|
||
|
||
|
||
@app.route('/admin/unblock/<ip>')
|
||
@admin_required
|
||
def admin_unblock(ip):
|
||
unblock_ip(ip)
|
||
database.log_security_event(
|
||
request.remote_addr,
|
||
'admin_unblock',
|
||
f'Ручная разблокировка IP {ip} администратором {session.get("username")}',
|
||
user_id=session.get('user_id')
|
||
)
|
||
flash(f'IP {ip} разблокирован', 'success')
|
||
return redirect(url_for('admin_panel'))
|
||
|
||
|
||
@app.route('/admin/block/<ip>')
|
||
@admin_required
|
||
def admin_block(ip):
|
||
block_ip(ip)
|
||
database.log_security_event(
|
||
request.remote_addr,
|
||
'manual_block',
|
||
f'Ручная блокировка IP {ip} администратором {session.get("username")}',
|
||
user_id=session.get('user_id')
|
||
)
|
||
flash(f'IP {ip} заблокирован', 'success')
|
||
return redirect(url_for('admin_panel'))
|
||
|
||
|
||
@app.route('/api/create', methods=['POST'])
|
||
@login_required
|
||
@limiter.limit("30 per minute")
|
||
def api_create_secret():
|
||
if request.content_length and request.content_length > 1024 * 1024: # 1MB
|
||
return jsonify({'error': 'Запрос слишком большой'}), 413
|
||
|
||
try:
|
||
data = request.get_json(force=True, silent=True)
|
||
if data is None:
|
||
return jsonify({'error': 'Неверный JSON'}), 400
|
||
except Exception as e:
|
||
app.logger.error(f"JSON parsing error: {e}")
|
||
return jsonify({'error': 'Неверный JSON'}), 400
|
||
|
||
if not data or 'secret' not in data:
|
||
return jsonify({'error': 'Поле "secret" обязательно'}), 400
|
||
|
||
secret_text = data['secret'].strip()
|
||
|
||
if not secret_text:
|
||
return jsonify({'error': 'Секрет не может быть пустым'}), 400
|
||
|
||
if len(secret_text) > app.config['MAX_SECRET_LENGTH']:
|
||
return jsonify({
|
||
'error': f'Секрет слишком длинный (макс. {app.config["MAX_SECRET_LENGTH"]})'
|
||
}), 400
|
||
|
||
if re.search(r'<script.*?>.*?</script>', secret_text, re.IGNORECASE | re.DOTALL):
|
||
return jsonify({'error': 'Обнаружено подозрительное содержимое'}), 400
|
||
|
||
client_info = get_client_info()
|
||
user_id = session.get('user_id')
|
||
|
||
if user_id:
|
||
user_exists = database.get_user_by_id(user_id)
|
||
if not user_exists:
|
||
session.clear()
|
||
return jsonify({'error': 'Сессия устарела. Войдите заново.'}), 401
|
||
|
||
try:
|
||
secret_id = store_secret(secret_text, client_info, user_id)
|
||
except Exception as e:
|
||
app.logger.error(f"Error creating secret via API: {e}")
|
||
return jsonify({'error': 'Ошибка при создании секрета'}), 500
|
||
|
||
view_url = url_for('view_secret', secret_id=secret_id, _external=True)
|
||
|
||
database.log_security_event(
|
||
client_info['ip'],
|
||
'api_secret_created',
|
||
f'Создан секрет через API: {secret_id} пользователем {session.get("username")}',
|
||
secret_id=secret_id,
|
||
user_id=user_id
|
||
)
|
||
|
||
return jsonify({
|
||
'status': 'ok',
|
||
'link': view_url,
|
||
'secret_id': secret_id,
|
||
'expires_in_hours': 24
|
||
})
|
||
|
||
|
||
@app.route('/api/check/<secret_id>')
|
||
def api_check_secret(secret_id):
|
||
info = get_secret_info(secret_id)
|
||
if not info:
|
||
return jsonify({'exists': False}), 404
|
||
log = database.get_secret_log(secret_id)
|
||
return jsonify({'exists': True, 'ttl_seconds': info['ttl_seconds'], 'ttl_hours': info['ttl_hours'],
|
||
'created_at': log['created_at'] if log else None, 'is_viewed': log['is_viewed'] if log else False})
|
||
|
||
|
||
@app.route('/api/status')
|
||
def api_status():
|
||
stats = database.get_stats()
|
||
return jsonify({'status': 'ok', 'statistics': stats, 'security': {'max_attempts': SECURITY_CONFIG['max_attempts'],
|
||
'block_duration': SECURITY_CONFIG[
|
||
'block_duration']}})
|
||
|
||
|
||
@app.route('/health')
|
||
def health_check():
|
||
try:
|
||
redis_ping = redis_client.ping()
|
||
database.db.session.execute('SELECT 1')
|
||
return jsonify({
|
||
'status': 'ok',
|
||
'redis': redis_ping,
|
||
'database': 'ok',
|
||
'timestamp': datetime.now().isoformat()
|
||
})
|
||
except Exception as e:
|
||
return jsonify({
|
||
'status': 'error',
|
||
'error': str(e),
|
||
'timestamp': datetime.now().isoformat()
|
||
}), 500
|
||
|
||
|
||
with app.app_context():
|
||
run_migrations()
|
||
create_admin()
|
||
|
||
if __name__ == '__main__':
|
||
app.run(
|
||
debug=app.config['DEBUG'],
|
||
host='0.0.0.0',
|
||
port=5000
|
||
)
|
||
|
||
app.jinja_env.tests['datetime'] = lambda value: isinstance(value, datetime)
|
||
app.jinja_env.tests['string'] = lambda value: isinstance(value, str)
|
||
app.jinja_env.tests['number'] = lambda value: isinstance(value, (int, float))
|
||
app.jinja_env.tests['list'] = lambda value: isinstance(value, list)
|
||
app.jinja_env.tests['dict'] = lambda value: isinstance(value, dict)
|