347 lines
11 KiB
Python
347 lines
11 KiB
Python
import os
|
|
from datetime import datetime, timedelta
|
|
from flask_sqlalchemy import SQLAlchemy
|
|
from sqlalchemy import Column, Integer, String, DateTime, Boolean, ForeignKey, Text
|
|
from sqlalchemy.orm import relationship
|
|
import bcrypt
|
|
import ipaddress
|
|
import html
|
|
|
|
db = SQLAlchemy()
|
|
|
|
|
|
class User(db.Model):
|
|
__tablename__ = 'users'
|
|
|
|
id = Column(Integer, primary_key=True)
|
|
username = Column(String(100), unique=True, nullable=False)
|
|
password_hash = Column(String(255), nullable=False)
|
|
role = Column(String(20), default='user')
|
|
created_at = Column(DateTime, default=datetime.now)
|
|
last_login = Column(DateTime, nullable=True)
|
|
is_active = Column(Boolean, default=True)
|
|
|
|
# ТОЛЬКО ЭТИ ПОЛЯ ДЛЯ ПРОФИЛЯ
|
|
email = Column(String(255), nullable=True)
|
|
jabber = Column(String(255), nullable=True)
|
|
updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)
|
|
|
|
# Настройки уведомлений
|
|
notify_email = Column(Boolean, default=False)
|
|
notify_jabber = Column(Boolean, default=False)
|
|
|
|
secrets = relationship('Secret', back_populates='user', cascade='all, delete-orphan')
|
|
security_logs = relationship('SecurityLog', back_populates='user', cascade='all, delete-orphan')
|
|
|
|
def set_password(self, password: str):
|
|
password_bytes = password.encode('utf-8')[:72]
|
|
salt = bcrypt.gensalt()
|
|
self.password_hash = bcrypt.hashpw(password_bytes, salt).decode('utf-8')
|
|
|
|
def check_password(self, password: str) -> bool:
|
|
password_bytes = password.encode('utf-8')[:72]
|
|
return bcrypt.checkpw(password_bytes, self.password_hash.encode('utf-8'))
|
|
|
|
def to_dict(self):
|
|
return {
|
|
'id': self.id,
|
|
'username': self.username,
|
|
'role': self.role,
|
|
'created_at': self.created_at.isoformat() if self.created_at else None,
|
|
'last_login': self.last_login.isoformat() if self.last_login else None,
|
|
'is_active': self.is_active,
|
|
'email': self.email,
|
|
'jabber': self.jabber,
|
|
'updated_at': self.updated_at.isoformat() if self.updated_at else None,
|
|
}
|
|
|
|
|
|
class Secret(db.Model):
|
|
__tablename__ = 'secrets'
|
|
|
|
id = Column(Integer, primary_key=True)
|
|
secret_id = Column(String(100), unique=True, nullable=False, index=True)
|
|
user_id = Column(Integer, ForeignKey('users.id', ondelete='CASCADE'), nullable=True)
|
|
created_at = Column(DateTime, default=datetime.now, index=True)
|
|
created_ip = Column(String(45), nullable=True)
|
|
created_user_agent = Column(String(255), nullable=True)
|
|
viewed_at = Column(DateTime, nullable=True)
|
|
viewed_ip = Column(String(45), nullable=True)
|
|
viewed_user_agent = Column(String(255), nullable=True)
|
|
is_viewed = Column(Boolean, default=False, index=True)
|
|
view_duration = Column(Integer, nullable=True)
|
|
request_note = Column(String(500), nullable=True)
|
|
|
|
user = relationship('User', back_populates='secrets')
|
|
security_logs = relationship('SecurityLog', back_populates='secret', cascade='all, delete-orphan')
|
|
|
|
def to_dict(self):
|
|
return {
|
|
'id': self.id,
|
|
'secret_id': self.secret_id,
|
|
'user_id': self.user_id,
|
|
'created_at': self.created_at.isoformat() if self.created_at else None,
|
|
'created_ip': self.created_ip,
|
|
'viewed_at': self.viewed_at.isoformat() if self.viewed_at else None,
|
|
'viewed_ip': self.viewed_ip,
|
|
'is_viewed': self.is_viewed,
|
|
'view_duration': self.view_duration,
|
|
'request_note': self.request_note
|
|
}
|
|
|
|
|
|
class SecurityLog(db.Model):
|
|
__tablename__ = 'security_logs'
|
|
|
|
id = Column(Integer, primary_key=True)
|
|
timestamp = Column(DateTime, default=datetime.now, index=True)
|
|
ip = Column(String(45), nullable=False, index=True)
|
|
event_type = Column(String(50), nullable=False, index=True)
|
|
details = Column(Text, nullable=True)
|
|
user_agent = Column(String(255), nullable=True)
|
|
secret_id = Column(String(100), ForeignKey('secrets.secret_id', ondelete='SET NULL'), nullable=True)
|
|
user_id = Column(Integer, ForeignKey('users.id', ondelete='SET NULL'), nullable=True)
|
|
|
|
user = relationship('User', back_populates='security_logs')
|
|
secret = relationship('Secret', back_populates='security_logs')
|
|
|
|
def to_dict(self):
|
|
return {
|
|
'id': self.id,
|
|
'timestamp': self.timestamp.isoformat() if self.timestamp else None,
|
|
'ip': self.ip,
|
|
'event_type': self.event_type,
|
|
'details': self.details,
|
|
'user_agent': self.user_agent,
|
|
'secret_id': self.secret_id,
|
|
'user_id': self.user_id
|
|
}
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
password_bytes = password.encode('utf-8')[:72]
|
|
salt = bcrypt.gensalt()
|
|
return bcrypt.hashpw(password_bytes, salt).decode('utf-8')
|
|
|
|
|
|
def verify_password(password: str, password_hash: str) -> bool:
|
|
password_bytes = password.encode('utf-8')[:72]
|
|
return bcrypt.checkpw(password_bytes, password_hash.encode('utf-8'))
|
|
|
|
|
|
def create_user(username: str, password: str, role: str = 'user') -> bool:
|
|
try:
|
|
existing = User.query.filter_by(username=username).first()
|
|
if existing:
|
|
return False
|
|
|
|
user = User(username=username, role=role)
|
|
user.set_password(password)
|
|
db.session.add(user)
|
|
db.session.commit()
|
|
return True
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
print(f"Ошибка создания пользователя: {e}")
|
|
return False
|
|
|
|
|
|
def get_user(username: str) -> User | None:
|
|
return User.query.filter_by(username=username).first()
|
|
|
|
|
|
def get_user_by_id(user_id: int) -> User | None:
|
|
return User.query.get(user_id)
|
|
|
|
|
|
def update_last_login(username: str):
|
|
user = get_user(username)
|
|
if user:
|
|
user.last_login = datetime.now()
|
|
db.session.commit()
|
|
|
|
|
|
def get_all_users() -> list:
|
|
return User.query.order_by(User.created_at.desc()).all()
|
|
|
|
|
|
def delete_user(user_id: int):
|
|
user = get_user_by_id(user_id)
|
|
if user:
|
|
db.session.delete(user)
|
|
db.session.commit()
|
|
|
|
|
|
def set_user_active(user_id: int, is_active: bool):
|
|
user = get_user_by_id(user_id)
|
|
if user:
|
|
user.is_active = is_active
|
|
db.session.commit()
|
|
|
|
|
|
def set_user_role(user_id: int, role: str):
|
|
user = get_user_by_id(user_id)
|
|
if user:
|
|
user.role = role
|
|
db.session.commit()
|
|
|
|
|
|
def log_creation(secret_id: str, client_info: dict, user_id: int = None, request_note: str = None, user_note: str = None):
|
|
secret = Secret(
|
|
secret_id=secret_id,
|
|
user_id=user_id,
|
|
created_ip=client_info.get('ip'),
|
|
created_user_agent=client_info.get('user_agent'),
|
|
request_note=request_note
|
|
)
|
|
db.session.add(secret)
|
|
db.session.commit()
|
|
|
|
|
|
def log_view(secret_id: str, client_info: dict):
|
|
secret = Secret.query.filter_by(secret_id=secret_id).first()
|
|
if secret:
|
|
if secret.created_at:
|
|
view_duration = int((datetime.now() - secret.created_at).total_seconds())
|
|
else:
|
|
view_duration = None
|
|
|
|
secret.viewed_at = datetime.now()
|
|
secret.viewed_ip = client_info.get('ip')
|
|
secret.viewed_user_agent = client_info.get('user_agent')
|
|
secret.is_viewed = True
|
|
secret.view_duration = view_duration
|
|
db.session.commit()
|
|
|
|
|
|
def log_security_event(ip: str, event_type: str, details: str = None,
|
|
user_agent: str = None, secret_id: str = None, user_id: int = None):
|
|
try:
|
|
ipaddress.ip_address(ip)
|
|
except ValueError:
|
|
ip = '0.0.0.0'
|
|
|
|
if details and len(details) > 1000:
|
|
details = details[:1000]
|
|
if user_agent:
|
|
user_agent = user_agent[:255]
|
|
|
|
log = SecurityLog(
|
|
ip=ip[:45],
|
|
event_type=event_type[:50],
|
|
details=details,
|
|
user_agent=user_agent,
|
|
secret_id=secret_id[:100] if secret_id else None,
|
|
user_id=user_id
|
|
)
|
|
|
|
try:
|
|
db.session.add(log)
|
|
db.session.commit()
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
print(f"Error logging security event: {e}")
|
|
|
|
|
|
def get_secret_log(secret_id: str) -> dict | None:
|
|
secret = Secret.query.filter_by(secret_id=secret_id).first()
|
|
return secret.to_dict() if secret else None
|
|
|
|
|
|
def get_all_logs(limit: int = 100, offset: int = 0, filter_viewed: str = None) -> list:
|
|
if limit < 1:
|
|
limit = 1
|
|
if limit > 1000:
|
|
limit = 1000
|
|
if offset < 0:
|
|
offset = 0
|
|
if offset > 100000:
|
|
offset = 100000
|
|
|
|
if filter_viewed not in ['viewed', 'pending', None]:
|
|
filter_viewed = None
|
|
|
|
query = Secret.query
|
|
if filter_viewed == 'viewed':
|
|
query = query.filter_by(is_viewed=True)
|
|
elif filter_viewed == 'pending':
|
|
query = query.filter_by(is_viewed=False)
|
|
|
|
return query.order_by(Secret.created_at.desc()).limit(limit).offset(offset).all()
|
|
|
|
|
|
def get_stats() -> dict:
|
|
total = Secret.query.count()
|
|
viewed = Secret.query.filter_by(is_viewed=True).count()
|
|
not_viewed = Secret.query.filter_by(is_viewed=False).count()
|
|
|
|
unique_creators = db.session.query(Secret.created_ip).distinct().count()
|
|
unique_viewers = db.session.query(Secret.viewed_ip).distinct().filter(Secret.viewed_ip.isnot(None)).count()
|
|
|
|
from sqlalchemy import func
|
|
avg_duration = db.session.query(func.avg(Secret.view_duration)).filter(Secret.is_viewed == True).scalar()
|
|
|
|
security_events = SecurityLog.query.count()
|
|
blocked_ips = SecurityLog.query.filter(
|
|
SecurityLog.event_type.in_(['temporary_block', 'permanent_block'])
|
|
).with_entities(SecurityLog.ip).distinct().count()
|
|
|
|
return {
|
|
'total': total,
|
|
'viewed': viewed,
|
|
'not_viewed': not_viewed,
|
|
'viewed_percent': round((viewed / total * 100) if total > 0 else 0, 2),
|
|
'unique_creators': unique_creators,
|
|
'unique_viewers': unique_viewers,
|
|
'avg_view_duration': round(avg_duration / 60, 2) if avg_duration else 0,
|
|
'security_events': security_events,
|
|
'blocked_ips': blocked_ips
|
|
}
|
|
|
|
|
|
def get_security_logs(limit: int = 50) -> list:
|
|
return SecurityLog.query.order_by(SecurityLog.timestamp.desc()).limit(limit).all()
|
|
|
|
|
|
def cleanup_old_data():
|
|
cutoff_date = datetime.now() - timedelta(days=30)
|
|
old_secrets = Secret.query.filter(Secret.created_at < cutoff_date).all()
|
|
for secret in old_secrets:
|
|
db.session.delete(secret)
|
|
|
|
cutoff_logs = datetime.now() - timedelta(days=90)
|
|
old_logs = SecurityLog.query.filter(SecurityLog.timestamp < cutoff_logs).all()
|
|
for log in old_logs:
|
|
db.session.delete(log)
|
|
|
|
db.session.commit()
|
|
|
|
|
|
def get_user_secrets(user_id: int, limit: int = 100) -> list:
|
|
if limit < 1:
|
|
limit = 1
|
|
if limit > 500:
|
|
limit = 500
|
|
|
|
secrets = Secret.query.filter_by(user_id=user_id).order_by(Secret.created_at.desc()).limit(limit).all()
|
|
|
|
for secret in secrets:
|
|
log = SecurityLog.query.filter_by(
|
|
secret_id=secret.secret_id,
|
|
event_type='secret_submitted_from_request'
|
|
).first()
|
|
secret.is_request = bool(log)
|
|
|
|
return secrets
|
|
|
|
|
|
def get_requests_for_user(user_id: int) -> list:
|
|
return SecurityLog.query.filter_by(
|
|
user_id=user_id,
|
|
event_type='secret_request_created'
|
|
).order_by(SecurityLog.timestamp.desc()).all()
|
|
|
|
|
|
def init_db():
|
|
db.create_all()
|