Logs_Kevorokan1
/
FuckingChat
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

MVC rewrite: config TUI, MariaDB support, Tenor GIFs, clean structure

This commit is contained in:
Your Name 2026-07-22 02:27:27 +03:00
parent 872052d265
commit 57257f6799
14 changed files with 743 additions and 303 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

116
README.md
View File

@ -1,53 +1,97 @@
# FuckingChat 🔒
Encrypted chat with channel support. Messages are encrypted at rest using Fernet (PBKDF2 + SHA256).
Encrypted chat with multi-channel support and Tenor GIFs.
Messages encrypted at rest (Fernet + PBKDF2/SHA256).
## Channels
- 💬 **General** — general discussion
- 🎮 **Games** — gaming chat
- 🎌 **Anime** — anime & manga
- ⚖️ **Politics** — politics discussion
## Setup
## Quick start
```bash
# Install dependencies
pip install bottle cryptography sqlalchemy
# Run (development)
python index.py
# First run — configure everything interactively
python app.py -c
# Run (production - recommended)
CHAT_SECRET="your-strong-secret" \
CHAT_SALT="your-random-salt" \
CHAT_ADMIN_TOKEN="your-admin-token" \
CHAT_RATE_LIMIT=5 \
CHAT_HOST=0.0.0.0 \
CHAT_PORT=8080 \
python index.py
# Then start the server
python app.py
```
## Environment Variables
## Project structure (MVC)
| Variable | Default | Description |
|---|---|---|
| `CHAT_SECRET` | `default_secret_change_me` | Encryption key |
| `CHAT_SALT` | `default_salt_change_me` | PBKDF2 salt |
| `CHAT_ADMIN_TOKEN` | (none) | Token for DELETE API |
| `CHAT_RATE_LIMIT` | `5` | Max messages per window |
| `CHAT_RATE_WINDOW` | `10` | Rate limit window (seconds) |
| `CHAT_MAX_MSG` | `2000` | Max message length |
| `CHAT_DB_PATH` | `./chat.db` | SQLite database path |
| `CHAT_HOST` | `localhost` | Listen address |
| `CHAT_PORT` | `8080` | Listen port |
| `CHAT_DEBUG` | `0` | Debug mode (set to `1` for dev) |
```
├── app.py # Entry point
├── index.html # Frontend (single-page)
├── config/
│ ├── __init__.py
│ ├── schema.py # Dataclass definitions
│ └── config.py # JSON serializer / loader
├── models/
│ ├── __init__.py
│ ├── base.py # SQLAlchemy ORM + engine factory
│ └── db.py # DBfrontend (encrypted CRUD)
├── controllers/
│ ├── __init__.py
│ ├── chat.py # Chat API endpoints
│ └── tenor.py # Tenor GIF proxy
├── views/
│ ├── __init__.py
│ └── config_tui.py # curses TUI configuration wizard
├── Anime/README.md
├── Games/README.md
├── Politics/README.md
└── README.md
```
## Security Features
## Configuration
Run `python app.py -c` to open the curses TUI:
- **Server** — host, port, debug mode
- **Security** — encryption secret, salt, admin token, rate limits
- **Database** — SQLite (default) or MariaDB
- **Tenor API** — API key for GIF search
Config is saved as `chat_config.json` in the project root.
Override path: `python app.py --config /path/to/config.json`
### Environment overrides
| Variable | Overrides config field |
|---|---|
| `CHAT_CONFIG` | Config file path |
| `CHAT_SECRET` | Encryption secret |
| `CHAT_SALT` | Encryption salt |
| `CHAT_ADMIN_TOKEN` | Admin token |
| `CHAT_RATE_LIMIT` | Rate limit |
| `CHAT_RATE_WINDOW` | Rate window (sec) |
| `CHAT_MAX_MSG` | Max message length |
| `CHAT_DB_PATH` | SQLite path |
| `CHAT_HOST` | Listen address |
| `CHAT_PORT` | Listen port |
| `CHAT_DEBUG` | Debug mode (1/0) |
| `TENOR_API_KEY` | Tenor API key |
## MariaDB setup
1. Choose **mariadb** as engine in config TUI.
2. Fill in host, port, user, password, database name.
3. Install driver: `pip install pymysql`
4. Start the server.
## Security
- ✅ Messages encrypted at rest (AES via Fernet)
- ✅ Rate limiting (configurable)
- ✅ IP hashing (privacy)
- ✅ XSS protection (HTML tag stripping)
- ✅ Admin token required for deletion
- ✅ XSS protection (HTML tag stripping, inline-gif whitelist)
- ✅ Admin token required for DELETE
- ✅ No debug mode in production
- ✅ API keys stored server-side only (Tenor proxy)
## API
- `GET /api/channels` — list channels
- `GET /api/messages?channel=X` — get messages
- `POST /api/messages` — send message `{"channel":"X","message":"..."}`
- `DELETE /api/messages/:id` — delete (requires `Authorization: Bearer <admin_token>`)
- `GET /api/gifs/trending` — trending GIFs
- `GET /api/gifs/search?q=...` — search GIFs

83
app.py Normal file
View File

@ -0,0 +1,83 @@
"""
app.py — FuckingChat entry point.
Usage:
python app.py # run server
python app.py --configure # run TUI configurator
python app.py --config path.json # specify config file
"""
import os
import sys
from bottle import Bottle, static_file, run
from config.config import load, save
from config.schema import AppConfig
from models.db import DBfrontend
from controllers.chat import init as chat_init, register_routes as chat_routes
from controllers.tenor import init as tenor_init, register_routes as tenor_routes
# Override config path from CLI
if "--config" in sys.argv:
idx = sys.argv.index("--config")
if idx + 1 < len(sys.argv):
os.environ["CHAT_CONFIG"] = sys.argv[idx + 1]
cfg: AppConfig = load()
def _make_db() -> DBfrontend:
return DBfrontend(
cfg.db,
secret=cfg.security.secret.encode(),
salt=cfg.security.salt.encode(),
)
def create_app() -> Bottle:
app = Bottle()
# Static frontend
static_dir = os.path.dirname(os.path.abspath(__file__))
@app.route("/")
def serve_index():
return static_file("index.html", root=static_dir)
@app.route("/static/<filename:path>")
def serve_static(filename):
return static_file(filename, root=os.path.join(static_dir, "static"))
# Init controllers
db = _make_db()
chat_init(db, cfg.security)
tenor_init(cfg.tenor)
# Register route blueprints
chat_routes(app)
tenor_routes(app)
return app
def run_server():
app = create_app()
host = cfg.server.host
port = cfg.server.port
debug = cfg.server.debug
if debug:
print("⚠️ DEBUG MODE — do not use in production!")
print(f"🚀 Chat running on http://{host}:{port}")
run(app=app, host=host, port=port, debug=debug, reloader=debug)
def run_configurator():
from views.config_tui import main
main()
if __name__ == "__main__":
if "--configure" in sys.argv or "-c" in sys.argv:
run_configurator()
else:
run_server()

1
config/__init__.py Normal file
View File

@ -0,0 +1 @@
# config

47
config/config.py Normal file
View File

@ -0,0 +1,47 @@
"""
config/config.py — config loader: file → AppConfig.
Stored as JSON in the same directory as the app.
"""
import json
import os
from pathlib import Path
from .schema import AppConfig, DatabaseConfig, SecurityConfig, ServerConfig, TenorConfig
CONFIG_FILE = os.environ.get("CHAT_CONFIG", "chat_config.json")
def _config_path() -> str:
"""Return absolute path to config file."""
p = Path(CONFIG_FILE)
if p.is_absolute():
return str(p)
return str(Path(__file__).resolve().parent.parent / CONFIG_FILE)
def load() -> AppConfig:
path = _config_path()
cfg = AppConfig()
if os.path.exists(path):
with open(path) as f:
data = json.load(f)
if "db" in data:
cfg.db = DatabaseConfig(**{**cfg.db.__dict__, **data["db"]})
if "security" in data:
cfg.security = SecurityConfig(**{**cfg.security.__dict__, **data["security"]})
if "server" in data:
cfg.server = ServerConfig(**{**cfg.server.__dict__, **data["server"]})
if "tenor" in data:
cfg.tenor = TenorConfig(**{**cfg.tenor.__dict__, **data["tenor"]})
return cfg
def save(cfg: AppConfig):
path = _config_path()
data = {
"db": cfg.db.__dict__,
"security": cfg.security.__dict__,
"server": cfg.server.__dict__,
"tenor": cfg.tenor.__dict__,
}
with open(path, "w") as f:
json.dump(data, f, indent=2)

42
config/schema.py Normal file
View File

@ -0,0 +1,42 @@
"""
config/schema.py — schema & defaults for chat configuration.
"""
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class DatabaseConfig:
engine: str = "sqlite" # "sqlite" | "mariadb"
sqlite_path: str = "chat.db"
mariadb_host: str = "localhost"
mariadb_port: int = 3306
mariadb_user: str = "chat"
mariadb_password: str = "chat"
mariadb_database: str = "chat"
@dataclass
class SecurityConfig:
secret: str = "default_secret_change_me"
salt: str = "default_salt_change_me"
admin_token: str = ""
rate_limit: int = 5
rate_window: int = 10
max_message_length: int = 2000
@dataclass
class ServerConfig:
host: str = "0.0.0.0"
port: int = 8080
debug: bool = False
@dataclass
class TenorConfig:
api_key: str = ""
@dataclass
class AppConfig:
db: DatabaseConfig = field(default_factory=DatabaseConfig)
security: SecurityConfig = field(default_factory=SecurityConfig)
server: ServerConfig = field(default_factory=ServerConfig)
tenor: TenorConfig = field(default_factory=TenorConfig)

1
controllers/__init__.py Normal file
View File

@ -0,0 +1 @@
# controllers/__init__.py

132
controllers/chat.py Normal file
View File

@ -0,0 +1,132 @@
"""
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 _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 = request.environ.get("REMOTE_ADDR", "127.0.0.1")
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()
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 = hashlib.sha256((_salt + ip).encode()).hexdigest()[:16]
msg_id = _db.add_message(channel, ip_hash, content)
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)}

52
controllers/tenor.py Normal file
View File

@ -0,0 +1,52 @@
"""
controllers/tenor.py — Tenor GIF API proxy.
"""
import json
import urllib.request
import urllib.parse
from bottle import route, request, response
_api_key: str = ""
def init(tenor_cfg):
global _api_key
_api_key = tenor_cfg.api_key
def register_routes(app):
@app.route("/api/gifs/trending", method="GET")
def gifs_trending():
if not _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={_api_key}&limit={limit}&media_filter=tinygif"
with urllib.request.urlopen(url) as resp:
data = json.loads(resp.read())
return {"success": True, "results": data.get("results", [])}
except Exception as e:
response.status = 502
return {"success": False, "error": str(e)}
@app.route("/api/gifs/search", method="GET")
def gifs_search():
if not _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={_api_key}&q={urllib.parse.quote(q)}&limit={limit}&media_filter=tinygif"
with urllib.request.urlopen(url) as resp:
data = json.loads(resp.read())
return {"success": True, "results": data.get("results", [])}
except Exception as e:
response.status = 502
return {"success": False, "error": str(e)}

267
index.py
View File

@ -1,267 +0,0 @@
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)

1
models/__init__.py Normal file
View File

@ -0,0 +1 @@
# models/__init__.py

34
models/base.py Normal file
View File

@ -0,0 +1,34 @@
"""
models/base.py — declarative base & the Messages model.
"""
from sqlalchemy import create_engine, Integer, String, Text
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker
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)
def make_engine(db_cfg):
"""Create a SQLAlchemy engine from DatabaseConfig."""
if db_cfg.engine == "mariadb":
url = (
f"mysql+pymysql://{db_cfg.mariadb_user}:{db_cfg.mariadb_password}"
f"@{db_cfg.mariadb_host}:{db_cfg.mariadb_port}/{db_cfg.mariadb_database}"
)
else:
url = f"sqlite:///{db_cfg.sqlite_path}"
engine = create_engine(url, echo=False)
Base.metadata.create_all(engine)
return engine

76
models/db.py Normal file
View File

@ -0,0 +1,76 @@
"""
models/db.py — DBfrontend: encrypted message store.
"""
import time
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
import base64
from sqlalchemy.orm import sessionmaker
from .base import make_engine, Messages
class DBfrontend:
def __init__(self, db_cfg, secret: bytes, salt: bytes):
self.engine = make_engine(db_cfg)
self.Session = sessionmaker(bind=self.engine)
self.cipher = self._get_cipher(secret, salt)
@staticmethod
def _get_cipher(secret: bytes, salt: bytes) -> Fernet:
kdf = PBKDF2HMAC(algorithm=hashes.SHA256(), length=32, salt=salt, iterations=100_000)
key = base64.urlsafe_b64encode(kdf.derive(secret))
return Fernet(key)
def _encrypt(self, message: str) -> str:
return self.cipher.encrypt(message.encode()).decode()
def _decrypt(self, encrypted: str) -> str:
try:
return self.cipher.decrypt(encrypted.encode()).decode()
except Exception:
return "[Decryption failed]"
def add_message(self, channel: str, ip: str, content: str) -> int:
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: str = "general", limit: int = 50) -> list:
with self.Session() as session:
q = (
session.query(Messages)
.filter(Messages.channel == channel)
.order_by(Messages.id.desc())
.limit(limit)
)
result = []
for msg in reversed(q.all()):
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: int) -> bool:
with self.Session() as session:
msg = session.get(Messages, msg_id)
if msg:
session.delete(msg)
session.commit()
return True
return False

1
views/__init__.py Normal file
View File

@ -0,0 +1 @@
# views/__init__.py

193
views/config_tui.py Normal file
View File

@ -0,0 +1,193 @@
"""
views/config_tui.py — curses-based configuration wizard.
Run: python -m views.config_tui
Or: python views/config_tui.py
"""
import curses
import sys
import os
# Ensure project root is on path when run directly
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from config.schema import AppConfig, DatabaseConfig, SecurityConfig, ServerConfig, TenorConfig
from config.config import load, save
# ── field descriptors ──────────────────────────────────────
SECTION_FIELDS = [
("server", "Server", [
("host", "Host", str, "0.0.0.0"),
("port", "Port", int, 8080),
("debug", "Debug mode", bool, False),
]),
("security", "Security", [
("secret", "Encryption secret", str, "default_secret_change_me"),
("salt", "Encryption salt", str, "default_salt_change_me"),
("admin_token", "Admin token", str, ""),
("rate_limit", "Rate limit (msg)", int, 5),
("rate_window", "Rate window (sec)", int, 10),
("max_message_length", "Max msg length", int, 2000),
]),
("db", "Database", [
("engine", "Engine (sqlite|mariadb)", str, "sqlite"),
("sqlite_path", "SQLite path", str, "chat.db"),
("mariadb_host", "MariaDB host", str, "localhost"),
("mariadb_port", "MariaDB port", int, 3306),
("mariadb_user", "MariaDB user", str, "chat"),
("mariadb_password", "MariaDB password", str, "chat"),
("mariadb_database", "MariaDB database", str, "chat"),
]),
("tenor", "Tenor API", [
("api_key", "API key", str, ""),
]),
]
# ── helpers ────────────────────────────────────────────────
def draw_menu(std, items, selected, top, height, width, title):
"""Draw a scrollable list and return the new top index."""
std.attron(curses.A_REVERSE)
std.addstr(0, 0, f" {title} ".center(width, "─"))
std.attroff(curses.A_REVERSE)
n = len(items)
visible = height - 2 # header + status line
# clamp top
if selected < top:
top = selected
if selected >= top + visible:
top = selected - visible + 1
top = max(0, min(top, n - visible))
for i in range(visible):
idx = top + i
if idx >= n:
break
line = f" {items[idx]}"
if idx == selected:
std.attron(curses.A_REVERSE)
std.addstr(1 + i, 0, line.ljust(width))
std.attroff(curses.A_REVERSE)
else:
std.addstr(1 + i, 0, line.ljust(width))
# status bar
std.attron(curses.A_REVERSE)
std.addstr(height - 1, 0, f" {selected+1}/{n} ↑↓ scroll ← back ".ljust(width))
std.attroff(curses.A_REVERSE)
return top
def edit_field(std, label, value, field_type):
"""Inline edit a field value. Returns new value or None if cancelled."""
h, w = std.getmaxyx()
prompt = f" {label}: "
# clear a couple lines
for y in range(2):
std.addstr(h // 2 + y, 0, " " * w)
std.addstr(h // 2, 0, prompt)
# pre-fill with current value
start = len(prompt)
curses.echo()
curses.curs_set(1)
raw = std.getstr(h // 2, start, w - start - 2).decode("utf-8", errors="replace").strip()
curses.noecho()
curses.curs_set(0)
if not raw:
# keep old value
return value
if field_type == bool:
return raw.lower() in ("1", "true", "yes", "y")
if field_type == int:
try:
return int(raw)
except ValueError:
return value
return raw
def run_config_tui(std, cfg: AppConfig) -> AppConfig | None:
curses.curs_set(0)
curses.use_default_colors()
h, w = std.getmaxyx()
if h < 12 or w < 50:
std.addstr(0, 0, "Terminal too small (min 50x12)")
std.refresh()
std.getch()
return None
# section list on the left
section_names = [s[1] for s in SECTION_FIELDS] # display names
section_data = [(s[0], s[2]) for s in SECTION_FIELDS] # (attr_name, fields)
sel_section = 0
sel_field = 0
top_section = 0
while True:
std.erase()
h, w = std.getmaxyx()
left_w = min(20, w // 3)
# draw sections
sname = section_names[sel_section]
draw_menu(std, section_names, sel_section, top_section, h, left_w, "Sections")
# draw field list on the right
sec_attr, fields = section_data[sel_section]
current_obj = getattr(cfg, sec_attr)
field_labels = [f"{f[1]}: {getattr(current_obj, f[0])}" for f in fields]
draw_menu(std, field_labels, sel_field, 0, h, w - left_w - 1, sname)
key = std.getch()
if key == curses.KEY_UP:
sel_field = max(0, sel_field - 1)
elif key == curses.KEY_DOWN:
sel_field = min(len(fields) - 1, sel_field + 1)
elif key == curses.KEY_LEFT:
sel_section = max(0, sel_section - 1)
sel_field = 0
top_section = 0
elif key == curses.KEY_RIGHT:
sel_section = min(len(section_names) - 1, sel_section + 1)
sel_field = 0
top_section = 0
elif key == ord("\n") or key == ord(" "):
# edit selected field
field_name, field_label, field_type, _ = fields[sel_field]
old = getattr(current_obj, field_name)
new = edit_field(std, field_label, old, field_type)
if new is not None:
setattr(current_obj, field_name, new)
elif key == ord("\t") or key == ord("s") or key == ord("S"):
# Save
save(cfg)
return cfg
elif key == ord("q") or key == 27: # ESC
return None
def main_std(std):
cfg = load()
result = run_config_tui(std, cfg)
if result:
save(result)
msg = " ✓ Config saved!"
else:
msg = " ✗ Cancelled."
h, w = std.getmaxyx()
std.addstr(h - 1, 0, msg.ljust(w - 1))
std.refresh()
std.getch()
def main():
curses.wrapper(main_std)
if __name__ == "__main__":
main()