268 lines
9.3 KiB
Python
268 lines
9.3 KiB
Python
from bottle import route, request, response, run, static_file, template
|
|
import hashlib
|
|
import time
|
|
import json
|
|
import os
|
|
import re
|
|
import urllib.request
|
|
import urllib.parse
|
|
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, Text
|
|
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
# --- Security config from environment ---
|
|
SALT = os.environ.get('CHAT_SALT', 'default_salt_change_me').encode()
|
|
PASSWORD = os.environ.get('CHAT_SECRET', 'default_secret_change_me').encode()
|
|
ADMIN_TOKEN = os.environ.get('CHAT_ADMIN_TOKEN', None)
|
|
RATE_LIMIT = int(os.environ.get('CHAT_RATE_LIMIT', 5)) # messages per window
|
|
RATE_WINDOW = int(os.environ.get('CHAT_RATE_WINDOW', 10)) # seconds
|
|
MAX_MSG_LENGTH = int(os.environ.get('CHAT_MAX_MSG', 2000))
|
|
DB_PATH = os.environ.get('CHAT_DB_PATH', os.path.join(os.path.dirname(os.path.abspath(__file__)), 'chat.db'))
|
|
TENOR_API_KEY = os.environ.get('TENOR_API_KEY', '')
|
|
|
|
# --- Rate limiting (in-memory) ---
|
|
rate_store = {}
|
|
|
|
def check_rate_limit(ip):
|
|
now = time.time()
|
|
window_start = now - RATE_WINDOW
|
|
if ip in rate_store:
|
|
rate_store[ip] = [t for t in rate_store[ip] if t > window_start]
|
|
if len(rate_store[ip]) >= RATE_LIMIT:
|
|
return False
|
|
rate_store[ip].append(now)
|
|
else:
|
|
rate_store[ip] = [now]
|
|
return True
|
|
|
|
# --- Encryption ---
|
|
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)
|
|
channel: Mapped[str] = mapped_column(String(32), nullable=False, default='general')
|
|
ip: Mapped[str] = mapped_column(String(64), nullable=False)
|
|
content: Mapped[str] = mapped_column(Text, nullable=False)
|
|
timestamp: Mapped[int] = mapped_column(Integer, nullable=False)
|
|
|
|
engine = create_engine(f"sqlite:///{DB_PATH}", echo=False)
|
|
Base.metadata.create_all(engine)
|
|
|
|
class DBfrontend():
|
|
def __init__(self):
|
|
self.Session = sessionmaker(bind=engine)
|
|
self.cipher = get_cipher()
|
|
|
|
def _encrypt(self, message):
|
|
return self.cipher.encrypt(message.encode()).decode()
|
|
|
|
def _decrypt(self, encrypted):
|
|
try:
|
|
return self.cipher.decrypt(encrypted.encode()).decode()
|
|
except:
|
|
return "[Decryption failed]"
|
|
|
|
def add_message(self, channel, ip, content):
|
|
with self.Session() as session:
|
|
msg = Messages(
|
|
channel=channel,
|
|
ip=ip,
|
|
content=self._encrypt(content),
|
|
timestamp=int(time.time() * 1000)
|
|
)
|
|
session.add(msg)
|
|
session.commit()
|
|
return msg.id
|
|
|
|
def get_messages(self, channel='general', limit=50):
|
|
with self.Session() as session:
|
|
q = session.query(Messages).filter(Messages.channel == channel)
|
|
q = q.order_by(Messages.id.desc()).limit(limit)
|
|
msgs = q.all()
|
|
result = []
|
|
for msg in reversed(msgs):
|
|
result.append({
|
|
'id': msg.id,
|
|
'channel': msg.channel,
|
|
'ip': msg.ip,
|
|
'content': self._decrypt(msg.content),
|
|
'timestamp': msg.timestamp
|
|
})
|
|
return result
|
|
|
|
def delete_message(self, msg_id):
|
|
with self.Session() as session:
|
|
msg = session.query(Messages).get(msg_id)
|
|
if msg:
|
|
session.delete(msg)
|
|
session.commit()
|
|
return True
|
|
return False
|
|
|
|
db = DBfrontend()
|
|
|
|
VALID_CHANNELS = {'general', 'games', 'anime', 'politics'}
|
|
CHANNEL_NAMES = {
|
|
'general': 'General',
|
|
'games': 'Games',
|
|
'anime': 'Anime',
|
|
'politics': 'Politics'
|
|
}
|
|
|
|
@route('/')
|
|
def serve_frontend():
|
|
return static_file('index.html', root=os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
@route('/api/channels', method='GET')
|
|
def list_channels():
|
|
response.content_type = 'application/json'
|
|
channels = [{'id': k, 'name': v} for k, v in CHANNEL_NAMES.items()]
|
|
return {'success': True, 'channels': channels}
|
|
|
|
@route('/api/messages', method='POST')
|
|
def post_message():
|
|
ip = request.environ.get('REMOTE_ADDR', '127.0.0.1')
|
|
|
|
if not check_rate_limit(ip):
|
|
response.status = 429
|
|
return {'success': False, 'error': 'Rate limit exceeded. Slow down.'}
|
|
|
|
try:
|
|
data = request.json
|
|
if not data:
|
|
response.status = 400
|
|
return {'success': False, 'error': 'No data provided'}
|
|
|
|
channel = data.get('channel', 'general')
|
|
content = data.get('message', '').strip()
|
|
|
|
if channel not in VALID_CHANNELS:
|
|
response.status = 400
|
|
return {'success': False, 'error': 'Invalid channel'}
|
|
|
|
if not content:
|
|
response.status = 400
|
|
return {'success': False, 'error': 'Empty message'}
|
|
|
|
if len(content) > MAX_MSG_LENGTH:
|
|
response.status = 400
|
|
return {'success': False, 'error': f'Message too long (max {MAX_MSG_LENGTH} chars)'}
|
|
|
|
# XSS filter - strip HTML tags except allowed inline-gif
|
|
allowed_gif = r'<img class="inline-gif" src="[^"]+">'
|
|
# Save allowed GIF tags before stripping
|
|
saved_gifs = re.findall(allowed_gif, content)
|
|
content = re.sub(r'<[^>]*>', '', content)
|
|
# Re-insert only saved GIFs at the end
|
|
if saved_gifs:
|
|
content = (content + '\n' + ''.join(saved_gifs)).strip()
|
|
|
|
ip_hash = hashlib.sha256((os.environ.get('CHAT_SALT', 'salt') + ip).encode()).hexdigest()[:16]
|
|
msg_id = db.add_message(channel, ip_hash, content)
|
|
|
|
response.content_type = 'application/json'
|
|
return {'success': True, 'message_id': msg_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:
|
|
channel = request.query.get('channel', 'general')
|
|
limit = min(int(request.query.get('limit', 50)), 200)
|
|
|
|
if channel not in VALID_CHANNELS:
|
|
channel = 'general'
|
|
|
|
messages = db.get_messages(channel=channel, limit=limit)
|
|
response.content_type = 'application/json'
|
|
return {'success': True, 'messages': messages}
|
|
|
|
except Exception as e:
|
|
response.status = 500
|
|
return {'success': False, 'error': str(e)}
|
|
|
|
@route('/api/gifs/trending', method='GET')
|
|
def gifs_trending():
|
|
if not TENOR_API_KEY:
|
|
response.status = 400
|
|
return {'success': False, 'error': 'Tenor API key not configured'}
|
|
try:
|
|
limit = request.query.get('limit', '20')
|
|
url = f'https://tenor.googleapis.com/v2/trending?key={TENOR_API_KEY}&limit={limit}&media_filter=tinygif'
|
|
with urllib.request.urlopen(url) as resp:
|
|
data = json.loads(resp.read())
|
|
response.content_type = 'application/json'
|
|
return {'success': True, 'results': data.get('results', [])}
|
|
except Exception as e:
|
|
response.status = 502
|
|
return {'success': False, 'error': str(e)}
|
|
|
|
@route('/api/gifs/search', method='GET')
|
|
def gifs_search():
|
|
if not TENOR_API_KEY:
|
|
response.status = 400
|
|
return {'success': False, 'error': 'Tenor API key not configured'}
|
|
try:
|
|
q = request.query.get('q', '')
|
|
if not q:
|
|
response.status = 400
|
|
return {'success': False, 'error': 'Missing query'}
|
|
limit = request.query.get('limit', '20')
|
|
url = f'https://tenor.googleapis.com/v2/search?key={TENOR_API_KEY}&q={urllib.parse.quote(q)}&limit={limit}&media_filter=tinygif'
|
|
with urllib.request.urlopen(url) as resp:
|
|
data = json.loads(resp.read())
|
|
response.content_type = 'application/json'
|
|
return {'success': True, 'results': data.get('results', [])}
|
|
except Exception as e:
|
|
response.status = 502
|
|
return {'success': False, 'error': str(e)}
|
|
|
|
@route('/api/messages/<message_id>', method='DELETE')
|
|
def delete_message(message_id):
|
|
# Admin token required for deletion
|
|
auth = request.get_header('Authorization', '')
|
|
token = auth.replace('Bearer ', '')
|
|
|
|
if ADMIN_TOKEN and token != ADMIN_TOKEN:
|
|
response.status = 403
|
|
return {'success': False, 'error': 'Forbidden'}
|
|
|
|
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__':
|
|
host = os.environ.get('CHAT_HOST', 'localhost')
|
|
port = int(os.environ.get('CHAT_PORT', 8080))
|
|
debug_mode = os.environ.get('CHAT_DEBUG', '0') == '1'
|
|
|
|
if debug_mode:
|
|
print(f"⚠️ WARNING: Debug mode is ON - do not use in production!")
|
|
|
|
run(host=host, port=port, debug=debug_mode)
|