1257 lines
44 KiB
Python
1257 lines
44 KiB
Python
import os
|
|
import base64
|
|
import secrets
|
|
import re
|
|
import bleach
|
|
from datetime import datetime
|
|
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:
|
|
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 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)
|
|
|
|
'''with app.app_context():
|
|
database.init_db()
|
|
create_admin()'''
|
|
|
|
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://"
|
|
)
|
|
|
|
|
|
@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}
|
|
|
|
|
|
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) -> 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_note = None
|
|
if request_note:
|
|
clean_note = sanitize_input(request_note)
|
|
if clean_note and len(clean_note) > 500:
|
|
clean_note = clean_note[:500]
|
|
|
|
database.log_creation(secret_id, client_info, user_id, clean_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('/create', methods=['POST'])
|
|
@login_required
|
|
@limiter.limit("30 per minute")
|
|
def create_secret():
|
|
secret_text = request.form.get('secret', '').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)
|
|
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), 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)
|
|
)
|
|
|
|
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 = []
|
|
|
|
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)
|
|
|
|
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': datetime.now()
|
|
})
|
|
|
|
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()
|
|
password = request.form.get('password', '')
|
|
confirm_password = request.form.get('confirm_password', '')
|
|
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 password or len(password) < 6:
|
|
flash('Пароль должен содержать минимум 6 символов', 'danger')
|
|
return render_template('admin_user_create.html')
|
|
|
|
if password != confirm_password:
|
|
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')
|
|
|
|
if database.create_user(username, password, role):
|
|
database.log_security_event(
|
|
request.remote_addr,
|
|
'user_created',
|
|
f'Создан пользователь: {username} (role: {role}) администратором {session.get("username")}',
|
|
user_id=session.get('user_id')
|
|
)
|
|
flash(f'Пользователь "{username}" успешно создан с ролью {role}', 'success')
|
|
return redirect(url_for('admin_panel'))
|
|
else:
|
|
flash('Ошибка при создании пользователя', 'danger')
|
|
|
|
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
|
|
|
|
|
|
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)
|