273 lines
9.3 KiB
Python
273 lines
9.3 KiB
Python
"""
|
|
controllers/chat.py — chat API endpoints.
|
|
"""
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import time
|
|
from bottle import route, request, response
|
|
|
|
from models.db import DBfrontend
|
|
|
|
VALID_CHANNELS = {"general", "games", "anime", "politics"}
|
|
CHANNEL_NAMES = {
|
|
"general": "General",
|
|
"games": "Games",
|
|
"anime": "Anime",
|
|
"politics": "Politics",
|
|
}
|
|
|
|
# In-memory rate store: ip → [timestamps]
|
|
rate_store: dict[str, list[float]] = {}
|
|
_rate_limit: int = 5
|
|
_rate_window: int = 10
|
|
_admin_token: str = ""
|
|
_salt: str = ""
|
|
_db: DBfrontend | None = None
|
|
|
|
|
|
def init(db: DBfrontend, sec_cfg):
|
|
global _db, _rate_limit, _rate_window, _admin_token, _salt
|
|
_db = db
|
|
_rate_limit = sec_cfg.rate_limit
|
|
_rate_window = sec_cfg.rate_window
|
|
_admin_token = sec_cfg.admin_token
|
|
_salt = sec_cfg.salt
|
|
|
|
|
|
def _get_client_ip() -> str:
|
|
forwarded = request.get_header("X-Forwarded-For")
|
|
if forwarded:
|
|
return forwarded.split(",")[0].strip()
|
|
real_ip = request.get_header("X-Real-IP")
|
|
if real_ip:
|
|
return real_ip.strip()
|
|
client_id = request.get_header("X-Client-Id")
|
|
if client_id:
|
|
return "client-" + client_id
|
|
return request.environ.get("REMOTE_ADDR", "127.0.0.1")
|
|
|
|
|
|
def _hash_ip(ip: str) -> str:
|
|
return hashlib.sha256((_salt + ip).encode()).hexdigest()[:16]
|
|
|
|
|
|
def _check_rate(ip: str) -> bool:
|
|
now = time.time()
|
|
cutoff = now - _rate_window
|
|
if ip in rate_store:
|
|
rate_store[ip] = [t for t in rate_store[ip] if t > cutoff]
|
|
if len(rate_store[ip]) >= _rate_limit:
|
|
return False
|
|
rate_store[ip].append(now)
|
|
else:
|
|
rate_store[ip] = [now]
|
|
return True
|
|
|
|
|
|
def _strip_xss(content: str) -> str:
|
|
"""Strip HTML tags except allowed <img class="inline-gif"> for Tenor."""
|
|
allowed = r'<img class="inline-gif" src="[^"]+">'
|
|
saved = re.findall(allowed, content)
|
|
content = re.sub(r"<[^>]*>", "", content)
|
|
if saved:
|
|
content = (content + "\n" + "".join(saved)).strip()
|
|
return content
|
|
|
|
|
|
def register_routes(app):
|
|
from bottle import Bottle
|
|
|
|
@app.route("/api/channels", method="GET")
|
|
def list_channels():
|
|
response.content_type = "application/json"
|
|
return {"success": True, "channels": [{"id": k, "name": v} for k, v in CHANNEL_NAMES.items()]}
|
|
|
|
@app.route("/api/messages", method="POST")
|
|
def post_message():
|
|
ip = _get_client_ip()
|
|
if not _check_rate(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"}
|
|
|
|
channel = data.get("channel", "general")
|
|
content = data.get("message", "").strip()
|
|
gif_url = data.get("gif_url") or None
|
|
|
|
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) > 2000:
|
|
response.status = 400
|
|
return {"success": False, "error": "Message too long"}
|
|
|
|
content = _strip_xss(content)
|
|
|
|
ip_hash = _hash_ip(ip)
|
|
msg_id = _db.add_message(channel, ip_hash, content, gif_url=gif_url)
|
|
|
|
return {"success": True, "message_id": msg_id, "ip": ip_hash}
|
|
except Exception as e:
|
|
response.status = 500
|
|
return {"success": False, "error": str(e)}
|
|
|
|
@app.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"
|
|
msgs = _db.get_messages(channel=channel, limit=limit)
|
|
return {"success": True, "messages": msgs}
|
|
except Exception as e:
|
|
response.status = 500
|
|
return {"success": False, "error": str(e)}
|
|
|
|
@app.route("/api/messages/<message_id:int>", method="DELETE")
|
|
def delete_message(message_id):
|
|
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:
|
|
ok = _db.delete_message(message_id)
|
|
return {"success": ok}
|
|
except Exception as e:
|
|
response.status = 500
|
|
return {"success": False, "error": str(e)}
|
|
|
|
@app.route("/api/polls", method="POST")
|
|
def create_poll():
|
|
ip = _get_client_ip()
|
|
if not _check_rate(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"}
|
|
|
|
channel = data.get("channel", "general")
|
|
question = data.get("question", "").strip()
|
|
options = data.get("options", [])
|
|
|
|
if channel not in VALID_CHANNELS:
|
|
response.status = 400
|
|
return {"success": False, "error": "Invalid channel"}
|
|
if not question:
|
|
response.status = 400
|
|
return {"success": False, "error": "Question is required"}
|
|
if len(question) > 500:
|
|
response.status = 400
|
|
return {"success": False, "error": "Question too long"}
|
|
if len(options) < 2:
|
|
response.status = 400
|
|
return {"success": False, "error": "At least 2 options required"}
|
|
if len(options) > 20:
|
|
response.status = 400
|
|
return {"success": False, "error": "Maximum 20 options allowed"}
|
|
for opt in options:
|
|
if not opt.strip():
|
|
response.status = 400
|
|
return {"success": False, "error": "Option cannot be empty"}
|
|
if len(opt) > 200:
|
|
response.status = 400
|
|
return {"success": False, "error": "Option too long"}
|
|
|
|
ip_hash = _hash_ip(ip)
|
|
expires_at = data.get("expires_at")
|
|
is_multiple = data.get("is_multiple", False)
|
|
is_anonymous = data.get("is_anonymous", False)
|
|
|
|
poll_id = _db.add_poll(
|
|
channel=channel,
|
|
creator_ip=ip_hash,
|
|
question=question,
|
|
options=options,
|
|
expires_at=expires_at,
|
|
is_multiple=is_multiple,
|
|
is_anonymous=is_anonymous,
|
|
)
|
|
return {"success": True, "poll_id": poll_id}
|
|
except Exception as e:
|
|
response.status = 500
|
|
return {"success": False, "error": str(e)}
|
|
|
|
@app.route("/api/polls", method="GET")
|
|
def list_polls():
|
|
try:
|
|
channel = request.query.get("channel", "general")
|
|
if channel not in VALID_CHANNELS:
|
|
channel = "general"
|
|
polls = _db.get_polls(channel=channel)
|
|
return {"success": True, "polls": polls}
|
|
except Exception as e:
|
|
response.status = 500
|
|
return {"success": False, "error": str(e)}
|
|
|
|
@app.route("/api/polls/<poll_id:int>", method="GET")
|
|
def get_poll(poll_id):
|
|
try:
|
|
poll = _db.get_poll(poll_id)
|
|
if not poll:
|
|
response.status = 404
|
|
return {"success": False, "error": "Poll not found"}
|
|
return {"success": True, "poll": poll}
|
|
except Exception as e:
|
|
response.status = 500
|
|
return {"success": False, "error": str(e)}
|
|
|
|
@app.route("/api/polls/<poll_id:int>/vote", method="POST")
|
|
def vote_poll(poll_id):
|
|
ip = _get_client_ip()
|
|
try:
|
|
data = request.json
|
|
if not data or "option_id" not in data:
|
|
response.status = 400
|
|
return {"success": False, "error": "option_id required"}
|
|
|
|
ip_hash = _hash_ip(ip)
|
|
ok, msg = _db.vote(poll_id, data["option_id"], ip_hash)
|
|
if not ok:
|
|
response.status = 400
|
|
return {"success": False, "error": msg}
|
|
return {"success": True, "message": msg}
|
|
except Exception as e:
|
|
response.status = 500
|
|
return {"success": False, "error": str(e)}
|
|
|
|
@app.route("/api/polls/<poll_id:int>/close", method="POST")
|
|
def close_poll(poll_id):
|
|
ip = _get_client_ip()
|
|
ip_hash = _hash_ip(ip)
|
|
|
|
auth = request.get_header("Authorization", "")
|
|
token = auth.replace("Bearer ", "")
|
|
if _admin_token and token == _admin_token:
|
|
ok, msg = _db.close_poll(poll_id, requester_ip=ip_hash)
|
|
if not ok:
|
|
response.status = 400
|
|
return {"success": False, "error": msg}
|
|
return {"success": True, "message": msg}
|
|
|
|
ok, msg = _db.close_poll(poll_id, requester_ip=ip_hash)
|
|
if not ok:
|
|
response.status = 400
|
|
return {"success": False, "error": msg}
|
|
return {"success": True, "message": msg}
|