240 lines
8.1 KiB
Python
240 lines
8.1 KiB
Python
from bottle import route, request, response, run, static_file
|
|
import hashlib
|
|
import time
|
|
import json
|
|
from cryptography.fernet import Fernet
|
|
from cryptography.hazmat.primitives import hashes
|
|
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
|
|
import base64
|
|
import sqlalchemy
|
|
from sqlalchemy import create_engine, String, Integer, DateTime
|
|
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
|
|
from sqlalchemy.orm import sessionmaker
|
|
from datetime import datetime
|
|
|
|
# Encryption setup
|
|
SALT = b'salt_123456789' # In production, use a random salt
|
|
PASSWORD = b'chat_secret_key_123' # In production, use environment variable
|
|
|
|
def get_cipher():
|
|
kdf = PBKDF2HMAC(
|
|
algorithm=hashes.SHA256(),
|
|
length=32,
|
|
salt=SALT,
|
|
iterations=100000,
|
|
)
|
|
key = base64.urlsafe_b64encode(kdf.derive(PASSWORD))
|
|
return Fernet(key)
|
|
|
|
class Base(DeclarativeBase):
|
|
pass
|
|
|
|
class Messages(Base):
|
|
__tablename__ = "messages"
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
|
ip: Mapped[str] = mapped_column(String, nullable=False)
|
|
content: Mapped[str] = mapped_column(String, nullable=False)
|
|
timestamp: Mapped[int] = mapped_column(Integer, nullable=False)
|
|
|
|
engine = create_engine("sqlite:///chat.db", echo=False)
|
|
Base.metadata.create_all(engine)
|
|
|
|
class DBfrontend():
|
|
def __init__(self):
|
|
self.Session = sessionmaker(bind=engine)
|
|
self.cipher = get_cipher()
|
|
|
|
def encrypt_message(self, message: str) -> str:
|
|
"""Encrypt message before storing"""
|
|
return self.cipher.encrypt(message.encode()).decode()
|
|
|
|
def decrypt_message(self, encrypted_message: str) -> str:
|
|
"""Decrypt message after retrieval"""
|
|
try:
|
|
return self.cipher.decrypt(encrypted_message.encode()).decode()
|
|
except:
|
|
return "[Decryption failed]"
|
|
|
|
def add_message(self, ip: str, content: str) -> int:
|
|
"""Add a new encrypted message to the database"""
|
|
with self.Session() as session:
|
|
encrypted_content = self.encrypt_message(content)
|
|
new_message = Messages(
|
|
ip=ip,
|
|
content=encrypted_content,
|
|
timestamp=int(time.time() * 1000)
|
|
)
|
|
session.add(new_message)
|
|
session.commit()
|
|
return new_message.id
|
|
|
|
def get_all_messages(self) -> list:
|
|
"""Get all messages and decrypt them"""
|
|
with self.Session() as session:
|
|
messages = session.query(Messages).all()
|
|
for msg in messages:
|
|
msg.content = self.decrypt_message(msg.content)
|
|
return messages
|
|
|
|
def get_message_by_id(self, message_id: int) -> Messages | None:
|
|
"""Get a specific message by its ID"""
|
|
with self.Session() as session:
|
|
message = session.query(Messages).get(message_id)
|
|
if message:
|
|
message.content = self.decrypt_message(message.content)
|
|
return message
|
|
|
|
def get_messages_by_ip(self, ip: str) -> list:
|
|
"""Get all messages from a specific IP address"""
|
|
with self.Session() as session:
|
|
messages = session.query(Messages).filter(Messages.ip == ip).all()
|
|
for msg in messages:
|
|
msg.content = self.decrypt_message(msg.content)
|
|
return messages
|
|
|
|
def update_message(self, message_id: int, new_content: str) -> bool:
|
|
"""Update the content of a message"""
|
|
with self.Session() as session:
|
|
message = session.query(Messages).get(message_id)
|
|
if message:
|
|
message.content = self.encrypt_message(new_content)
|
|
session.commit()
|
|
return True
|
|
return False
|
|
|
|
def delete_message(self, message_id: int) -> bool:
|
|
"""Delete a message by its ID"""
|
|
with self.Session() as session:
|
|
message = session.query(Messages).get(message_id)
|
|
if message:
|
|
session.delete(message)
|
|
session.commit()
|
|
return True
|
|
return False
|
|
|
|
def delete_messages_by_ip(self, ip: str) -> int:
|
|
"""Delete all messages from a specific IP address"""
|
|
with self.Session() as session:
|
|
messages = session.query(Messages).filter(Messages.ip == ip).all()
|
|
count = len(messages)
|
|
for message in messages:
|
|
session.delete(message)
|
|
session.commit()
|
|
return count
|
|
|
|
def get_message_count(self) -> int:
|
|
"""Get total number of messages"""
|
|
with self.Session() as session:
|
|
return session.query(Messages).count()
|
|
|
|
def get_latest_messages(self, limit: int = 10) -> list:
|
|
"""Get the latest N messages"""
|
|
with self.Session() as session:
|
|
messages = session.query(Messages).order_by(Messages.id.desc()).limit(limit).all()
|
|
for msg in messages:
|
|
msg.content = self.decrypt_message(msg.content)
|
|
return messages
|
|
|
|
# Initialize database
|
|
db = DBfrontend()
|
|
|
|
# Routes
|
|
import os
|
|
|
|
@route('/')
|
|
def serve_frontend():
|
|
# Получаем путь к директории, где находится index.py
|
|
current_dir = os.path.dirname(os.path.abspath(__file__))
|
|
return static_file('index.html', root=current_dir)
|
|
|
|
@route('/api/messages', method='POST')
|
|
def post_message():
|
|
ip = request.environ.get('REMOTE_ADDR', '127.0.0.1')
|
|
|
|
try:
|
|
data = request.json
|
|
if not data or 'message' not in data:
|
|
response.status = 400
|
|
return {'success': False, 'error': 'No message provided'}
|
|
|
|
message = data.get('message', '').strip()
|
|
if not message:
|
|
response.status = 400
|
|
return {'success': False, 'error': 'Empty message'}
|
|
|
|
# Get IP hash for privacy
|
|
ip_hash = hashlib.sha256(('salt' + ip).encode('utf-8')).hexdigest()[:16]
|
|
|
|
# Save message with encrypted content
|
|
message_id = db.add_message(ip_hash, message)
|
|
|
|
response.content_type = 'application/json'
|
|
return {
|
|
'success': True,
|
|
'message_id': message_id,
|
|
'ip': ip_hash
|
|
}
|
|
|
|
except Exception as e:
|
|
response.status = 500
|
|
return {'success': False, 'error': str(e)}
|
|
|
|
@route('/api/messages', method='GET')
|
|
def get_messages():
|
|
try:
|
|
limit = int(request.query.get('limit', 20))
|
|
messages = db.get_latest_messages(limit)
|
|
|
|
result = []
|
|
for msg in messages:
|
|
result.append({
|
|
'id': msg.id,
|
|
'ip': msg.ip,
|
|
'content': msg.content,
|
|
'timestamp': msg.timestamp
|
|
})
|
|
|
|
response.content_type = 'application/json'
|
|
return {'success': True, 'messages': result}
|
|
|
|
except Exception as e:
|
|
response.status = 500
|
|
return {'success': False, 'error': str(e)}
|
|
|
|
@route('/api/messages/<message_id>', method='PUT')
|
|
def update_message(message_id):
|
|
try:
|
|
data = request.json
|
|
if not data or 'message' not in data:
|
|
response.status = 400
|
|
return {'success': False, 'error': 'No message provided'}
|
|
|
|
new_content = data.get('message', '').strip()
|
|
if not new_content:
|
|
response.status = 400
|
|
return {'success': False, 'error': 'Empty message'}
|
|
|
|
success = db.update_message(int(message_id), new_content)
|
|
|
|
response.content_type = 'application/json'
|
|
return {'success': success}
|
|
|
|
except Exception as e:
|
|
response.status = 500
|
|
return {'success': False, 'error': str(e)}
|
|
|
|
@route('/api/messages/<message_id>', method='DELETE')
|
|
def delete_message(message_id):
|
|
try:
|
|
success = db.delete_message(int(message_id))
|
|
|
|
response.content_type = 'application/json'
|
|
return {'success': success}
|
|
|
|
except Exception as e:
|
|
response.status = 500
|
|
return {'success': False, 'error': str(e)}
|
|
|
|
if __name__ == '__main__':
|
|
run(host='localhost', port=8080, debug=True) |