Update index.py
This commit is contained in:
parent
ca1884403e
commit
b1d179c4c5
241
index.py
241
index.py
|
|
@ -1,240 +1 @@
|
|||
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)
|
||||
ZnJvbSBmbGFzayBtZXNzYWdlX2NoYW5uZWwgYWQtY2hhbm5lbHMtdXBkCmNsYXNzICI8c29jaWFsLXBvbGl0
|
||||
Loading…
Reference in New Issue