Forge Paste

#!/usr/bin/env python3 """ Forge Paste — A lightweight, self-hosted paste server. Hastebin-compatible API with a clean web UI. Usage: python3 forge-paste.py [--port PORT] [--host HOST] [--db PATH] [--max-age DAYS] API: POST /api/paste — raw text body → {"key": "abc123", "url": "..."} POST /api/documents — hastebin-compat → {"key": "abc123"} GET /api/raw/ — raw text output GET /api/documents/ — {"key": "...", "data": "..."} GET / — web viewer GET / — web editor """ import sqlite3 import string import random import os import json import argparse import time from http.server import HTTPServer, BaseHTTPRequestHandler from urllib.parse import urlparse from datetime import datetime import html DB_PATH = os.environ.get("FORGE_PASTE_DB", "/var/lib/forge-paste/pastes.db") HOST = os.environ.get("FORGE_PASTE_HOST", "127.0.0.1") PORT = int(os.environ.get("FORGE_PASTE_PORT", "7890")) MAX_SIZE = 512 * 1024 # 512KB MAX_AGE_DAYS = int(os.environ.get("FORGE_PASTE_MAX_AGE", "30")) # Default 30 days _CLEANUP_INTERVAL = 3600 # Run cleanup at most once per hour _last_cleanup = 0 def get_db(): """Get a thread-local database connection.""" db = sqlite3.connect(DB_PATH) db.row_factory = sqlite3.Row db.execute("PRAGMA journal_mode=WAL") db.execute("PRAGMA foreign_keys=ON") return db def init_db(): """Initialize the database schema.""" os.makedirs(os.path.dirname(DB_PATH), exist_ok=True) db = get_db() db.execute(""" CREATE TABLE IF NOT EXISTS pastes ( id INTEGER PRIMARY KEY AUTOINCREMENT, slug TEXT UNIQUE NOT NULL, content TEXT NOT NULL, language TEXT, created_at REAL NOT NULL DEFAULT (unixepoch()), views INTEGER NOT NULL DEFAULT 0, ip TEXT, expires_at REAL ) """) # Migrate: add expires_at column if missing (existing installs) cols = [row[1] for row in db.execute("PRAGMA table_info(pastes)").fetchall()] if "expires_at" not in cols: db.execute("ALTER TABLE pastes ADD COLUMN expires_at REAL") # Backfill existing pastes: expire them MAX_AGE_DAYS from creation db.execute( "UPDATE pastes SET expires_at = created_at + ? WHERE expires_at IS NULL", (MAX_AGE_DAYS * 86400,), ) db.commit() db.execute("CREATE INDEX IF NOT EXISTS idx_slug ON pastes(slug)") db.execute("CREATE INDEX IF NOT EXISTS idx_created ON pastes(created_at)") db.execute("CREATE INDEX IF NOT EXISTS idx_expires ON pastes(expires_at)") db.commit() db.close() def generate_slug(length=8): chars = string.ascii_lowercase + string.digits return ''.join(random.choices(chars, k=length)) def cleanup_expired(): """Delete expired pastes. Runs at most once per _CLEANUP_INTERVAL.""" global _last_cleanup now = time.time() if now - _last_cleanup < _CLEANUP_INTERVAL: return _last_cleanup = now db = get_db() deleted = db.execute( "DELETE FROM pastes WHERE expires_at IS NOT NULL AND expires_at < ?", (now,) ).rowcount if deleted: db.execute("PRAGMA wal_checkpoint(TRUNCATE)") db.commit() db.close() def create_paste(content, language=None, ip=None): cleanup_expired() # Opportunistic cleanup db = get_db() slug = generate_slug() # Ensure unique while db.execute("SELECT 1 FROM pastes WHERE slug=?", (slug,)).fetchone(): slug = generate_slug() now = time.time() expires_at = now + (MAX_AGE_DAYS * 86400) if MAX_AGE_DAYS > 0 else None db.execute( "INSERT INTO pastes (slug, content, language, created_at, views, ip, expires_at) VALUES (?, ?, ?, ?, 0, ?, ?)", (slug, content, language, now, ip, expires_at) ) db.commit() db.close() return slug def get_paste(slug): db = get_db() # Exclude expired pastes from reads row = db.execute( "SELECT * FROM pastes WHERE slug=? AND (expires_at IS NULL OR expires_at > ?)", (slug, time.time()), ).fetchone() if row: db.execute("UPDATE pastes SET views = views + 1 WHERE slug=?", (slug,)) db.commit() db.close() return dict(row) if row else None def get_stats(): db = get_db() row = db.execute("SELECT COUNT(*) as total FROM pastes").fetchone() db.close() return row["total"] def time_ago(ts): seconds = int(time.time() - ts) if seconds < 60: return "just now" if seconds < 3600: return f"{seconds // 60}m ago" if seconds < 86400: return f"{seconds // 3600}h ago" return f"{seconds // 86400}d ago" # ── HTML Templates ────────────────────────────────────────── STYLE = """ """ ICON_SVG = '' def page_editor(host): total = get_stats() return f""" Forge Paste{STYLE}

Forge Paste — API: curl -X POST -d "your text" {html.escape(host)}/api/paste

""" def page_viewer(paste, host, is_new=False): lines = paste["content"].split("\n") gw = len(str(len(lines))) line_html = "" for i, line in enumerate(lines, 1): line_html += f'

{i}{html.escape(line)}

' ta = time_ago(paste["created_at"]) v = paste["views"] slug = paste["slug"] url = f"{host}/{slug}" expires_at = paste.get("expires_at") if expires_at: expires_in = expires_at - time.time() if expires_in > 86400: expiry_text = f"expires in {int(expires_in // 86400)}d" elif expires_in > 3600: expiry_text = f"expires in {int(expires_in // 3600)}h" elif expires_in > 0: expiry_text = "expires soon" else: expiry_text = "expired" else: expiry_text = "no expiry" if is_new: return f""" {slug} — Forge Paste{STYLE}

✓ Paste created!

{html.escape(url)}

View Paste New Paste

""" return f""" {slug} — Forge Paste{STYLE}

{line_html}

{html.escape(paste["content"])}

""" def page_notfound(): return f""" Not Found — Forge Paste{STYLE}

🔍

Paste not found

This paste may have expired or never existed.

Create New Paste

""" class PasteHandler(BaseHTTPRequestHandler): def log_message(self, format, *args): # Quiet logging pass def _send(self, code, content, content_type="text/html"): self.send_response(code) self.send_header("Content-Type", content_type) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") self.end_headers() if isinstance(content, str): content = content.encode("utf-8") self.wfile.write(content) def _host(self): proto = self.headers.get("X-Forwarded-Proto", "http") host = self.headers.get("X-Forwarded-Host") or self.headers.get("Host", "localhost") return f"{proto}://{host}" def do_OPTIONS(self): self._send(204, "") def do_GET(self): parsed = urlparse(self.path) path = parsed.path.rstrip("/") query = parsed.query host = self._host() if path == "" or path == "/": return self._send(200, page_editor(host)) # API routes if path.startswith("/api/raw/"): slug = path[9:] paste = get_paste(slug) if not paste: return self._send(404, "Not found", "text/plain") return self._send(200, paste["content"], "text/plain; charset=utf-8") if path.startswith("/api/documents/"): slug = path[15:] paste = get_paste(slug) if not paste: return self._send(404, json.dumps({"error": "not found"}), "application/json") return self._send(200, json.dumps({"key": paste["slug"], "data": paste["content"]}), "application/json") # Web viewer slug = path[1:] # Remove leading / if slug and all(c in string.ascii_lowercase + string.digits for c in slug): paste = get_paste(slug) if paste: is_new = "new=1" in query return self._send(200, page_viewer(paste, host, is_new)) return self._send(404, page_notfound()) self._send(404, page_notfound()) def do_POST(self): try: length = int(self.headers.get("Content-Length", 0)) except (ValueError, TypeError): return self._send(400, json.dumps({"error": "Invalid Content-Length"}), "application/json") if length < 0: return self._send(400, json.dumps({"error": "Invalid Content-Length"}), "application/json") if length > MAX_SIZE: return self._send(413, json.dumps({"error": "Content too large (max 512KB)"}), "application/json") body = self.rfile.read(length).decode("utf-8", errors="replace") if not body.strip(): return self._send(400, json.dumps({"error": "Empty content"}), "application/json") parsed = urlparse(self.path) path = parsed.path.rstrip("/") host = self._host() ip = self.headers.get("X-Real-IP") or self.client_address[0] slug = create_paste(body, ip=ip) if path == "/api/documents": # Hastebin-compatible response return self._send(200, json.dumps({"key": slug}), "application/json") # Default API response self._send(200, json.dumps({"key": slug, "url": f"{host}/{slug}"}), "application/json") def main(): global DB_PATH, HOST, PORT, MAX_AGE_DAYS parser = argparse.ArgumentParser(description="Forge Paste Server") parser.add_argument("--port", type=int, default=PORT, help=f"Port (default: {PORT})") parser.add_argument("--host", default=HOST, help=f"Host (default: {HOST})") parser.add_argument("--db", default=DB_PATH, help=f"Database path (default: {DB_PATH})") parser.add_argument("--max-age", type=int, default=MAX_AGE_DAYS, help=f"Paste TTL in days, 0 = never expire (default: {MAX_AGE_DAYS})") args = parser.parse_args() DB_PATH = args.db HOST = args.host PORT = args.port MAX_AGE_DAYS = args.max_age init_db() server = HTTPServer((HOST, PORT), PasteHandler) print(f"🔥 Forge Paste running on http://{HOST}:{PORT}") print(f"📦 Database: {DB_PATH}") try: server.serve_forever() except KeyboardInterrupt: print("\nShutting down...") server.server_close() if __name__ == "__main__": main()