#!/usr/bin/env python3 """ Forge Paste — The ultimate self-hosted pastebin. Zero dependencies · Hastebin-compatible · Full REST API · Syntax highlighting Usage: python3 forge-paste.py [--port PORT] [--host HOST] [--db PATH] [--max-age DAYS] API: POST /api/paste — Create paste (JSON body or raw text) GET /api/paste/ — Get paste (JSON) PUT /api/paste/ — Update paste (requires edit_token) DELETE /api/paste/ — Delete paste (requires edit_token) GET /api/raw/ — Raw text output GET /api/download/ — Download as file GET /api/pastes/recent — List recent pastes GET /api/stats — Server stats GET /api/languages — Supported language list GET /api/health — Health check POST /api/documents — Hastebin-compat create GET /api/documents/ — Hastebin-compat get """ import sqlite3, string, random, os, json, argparse, time, hashlib, secrets, re, gzip from http.server import HTTPServer, BaseHTTPRequestHandler from urllib.parse import urlparse, parse_qs from datetime import datetime from io import BytesIO from threading import Thread import html as html_mod # ── Configuration ─────────────────────────────────────────── 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 = 2 * 1024 * 1024 # 2MB MAX_AGE_DAYS = int(os.environ.get("FORGE_PASTE_MAX_AGE", "30")) _CLEANUP_INTERVAL = 3600 _last_cleanup = 0 VERSION = "2.0.0" # ── Expiry presets (seconds) ──────────────────────────────── EXPIRY_PRESETS = { "10m": 600, "1h": 3600, "1d": 86400, "7d": 604800, "30d": 2592000, "90d": 7776000, "1y": 31536000, "never": 0, } # ── Languages ─────────────────────────────────────────────── LANGUAGES = [ "plaintext","bash","c","cpp","csharp","css","dart","diff","dockerfile", "elixir","erlang","go","graphql","haskell","html","ini","java", "javascript","json","julia","kotlin","latex","lua","makefile","markdown", "nginx","objectivec","ocaml","perl","php","powershell","properties", "python","r","ruby","rust","scala","scss","shell","sql","swift", "toml","typescript","xml","yaml","zig", ] # Extensions → language mapping for auto-detect EXT_MAP = { "py":"python","js":"javascript","ts":"typescript","rb":"ruby","rs":"rust", "go":"go","java":"java","c":"c","cpp":"cpp","h":"c","hpp":"cpp", "cs":"csharp","php":"php","sh":"bash","bash":"bash","zsh":"bash", "sql":"sql","html":"html","htm":"html","css":"css","scss":"scss", "json":"json","xml":"xml","yaml":"yaml","yml":"yaml","toml":"toml", "md":"markdown","dockerfile":"dockerfile","makefile":"makefile", "lua":"lua","swift":"swift","kt":"kotlin","scala":"scala", "hs":"haskell","ex":"elixir","erl":"erlang","r":"r","jl":"julia", "dart":"dart","zig":"zig","ini":"ini","conf":"ini","cfg":"ini", "diff":"diff","patch":"diff","ps1":"powershell", } def detect_language(content, filename=None): """Auto-detect language from filename extension or content heuristics.""" if filename: ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else "" if ext in EXT_MAP: return EXT_MAP[ext] # Content heuristics first = content[:200].strip() if first.startswith("#!/usr/bin/env python") or first.startswith("#!/usr/bin/python"): return "python" if first.startswith("#!/bin/bash") or first.startswith("#!/bin/sh"): return "bash" if first.startswith("#!/usr/bin/env node"): return "javascript" if first.startswith(" 1000 else content) return "json" except Exception: pass if re.match(r"^(SELECT|INSERT|UPDATE|DELETE|CREATE|ALTER|DROP)\s", first, re.I): return "sql" if re.match(r"^(FROM|RUN|CMD|COPY|ENV|EXPOSE|WORKDIR)\s", first, re.I): return "dockerfile" if "def " in first and ":" in first: return "python" if "function " in first or "const " in first or "let " in first: return "javascript" if re.match(r"^(diff|---|\+\+\+|@@)\s", first): return "diff" return "plaintext" # ── Database ──────────────────────────────────────────────── def get_db(): 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(): 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, title TEXT, content TEXT NOT NULL, language TEXT, created_at REAL NOT NULL DEFAULT (unixepoch()), updated_at REAL, views INTEGER NOT NULL DEFAULT 0, ip TEXT, expires_at REAL, burn_after_read INTEGER NOT NULL DEFAULT 0, edit_token TEXT, parent_slug TEXT, size INTEGER NOT NULL DEFAULT 0, is_public INTEGER NOT NULL DEFAULT 1 ) """) # Migrations for existing installs cols = [row[1] for row in db.execute("PRAGMA table_info(pastes)").fetchall()] migrations = { "expires_at": "ALTER TABLE pastes ADD COLUMN expires_at REAL", "burn_after_read": "ALTER TABLE pastes ADD COLUMN burn_after_read INTEGER NOT NULL DEFAULT 0", "edit_token": "ALTER TABLE pastes ADD COLUMN edit_token TEXT", "parent_slug": "ALTER TABLE pastes ADD COLUMN parent_slug TEXT", "size": "ALTER TABLE pastes ADD COLUMN size INTEGER NOT NULL DEFAULT 0", "is_public": "ALTER TABLE pastes ADD COLUMN is_public INTEGER NOT NULL DEFAULT 1", "title": "ALTER TABLE pastes ADD COLUMN title TEXT", "updated_at": "ALTER TABLE pastes ADD COLUMN updated_at REAL", } for col, sql in migrations.items(): if col not in cols: db.execute(sql) # Backfill size db.execute("UPDATE pastes SET size = LENGTH(content) WHERE size = 0 AND content IS NOT NULL") # Backfill expires_at if MAX_AGE_DAYS > 0: 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.execute("CREATE INDEX IF NOT EXISTS idx_public ON pastes(is_public)") 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(): global _last_cleanup now = time.time() if now - _last_cleanup < _CLEANUP_INTERVAL: return _last_cleanup = now db = get_db() # Delete expired db.execute("DELETE FROM pastes WHERE expires_at IS NOT NULL AND expires_at < ?", (now,)) db.commit() db.close() def create_paste(content, title=None, language=None, ip=None, expires_in=None, burn=False, is_public=True, parent_slug=None): cleanup_expired() db = get_db() slug = generate_slug() while db.execute("SELECT 1 FROM pastes WHERE slug=?", (slug,)).fetchone(): slug = generate_slug() edit_token = secrets.token_urlsafe(24) now = time.time() if expires_in is not None: expires_at = now + expires_in if expires_in > 0 else None elif MAX_AGE_DAYS > 0: expires_at = now + (MAX_AGE_DAYS * 86400) else: expires_at = None if not language: language = detect_language(content) size = len(content.encode("utf-8")) db.execute( """INSERT INTO pastes (slug, title, content, language, created_at, views, ip, expires_at, burn_after_read, edit_token, parent_slug, size, is_public) VALUES (?, ?, ?, ?, ?, 0, ?, ?, ?, ?, ?, ?, ?)""", (slug, title, content, language, now, ip, expires_at, 1 if burn else 0, edit_token, parent_slug, size, 1 if is_public else 0) ) db.commit() db.close() return slug, edit_token def get_paste(slug, increment_view=True): db = get_db() row = db.execute( "SELECT * FROM pastes WHERE slug=? AND (expires_at IS NULL OR expires_at > ?)", (slug, time.time()), ).fetchone() if not row: db.close() return None paste = dict(row) if increment_view: db.execute("UPDATE pastes SET views = views + 1 WHERE slug=?", (slug,)) db.commit() # Handle burn after read if paste.get("burn_after_read") and increment_view: db.execute("DELETE FROM pastes WHERE slug=?", (slug,)) db.commit() paste["_burned"] = True db.close() return paste def update_paste(slug, content, title=None, language=None): db = get_db() now = time.time() size = len(content.encode("utf-8")) if not language: language = detect_language(content) db.execute( "UPDATE pastes SET content=?, title=?, language=?, updated_at=?, size=? WHERE slug=?", (content, title, language, now, size, slug) ) db.commit() db.close() def delete_paste(slug): db = get_db() db.execute("DELETE FROM pastes WHERE slug=?", (slug,)) db.commit() db.close() def get_recent(limit=20, offset=0): db = get_db() rows = db.execute( """SELECT slug, title, language, created_at, views, size, burn_after_read, expires_at FROM pastes WHERE is_public=1 AND (expires_at IS NULL OR expires_at > ?) ORDER BY created_at DESC LIMIT ? OFFSET ?""", (time.time(), limit, offset) ).fetchall() db.close() return [dict(r) for r in rows] def get_stats(): db = get_db() total = db.execute("SELECT COUNT(*) FROM pastes").fetchone()[0] total_views = db.execute("SELECT COALESCE(SUM(views),0) FROM pastes").fetchone()[0] total_size = db.execute("SELECT COALESCE(SUM(size),0) FROM pastes").fetchone()[0] active = db.execute( "SELECT COUNT(*) FROM pastes WHERE expires_at IS NULL OR expires_at > ?", (time.time(),)).fetchone()[0] db.close() return {"total_pastes": total, "active_pastes": active, "total_views": total_views, "total_size": total_size, "version": VERSION} 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" def format_bytes(b): if b < 1024: return f"{b} B" if b < 1048576: return f"{b/1024:.1f} KB" return f"{b/1048576:.1f} MB" # ── HTML Templates ────────────────────────────────────────── ICON_SVG = '' FAVICON = "data:image/svg+xml," CSS = """ """ HLJS_HEAD = """ """ TOAST_JS = """
""" def page_head(title, extra_head=""): return f""" {html_mod.escape(title)} {CSS}{extra_head}""" def page_nav(slug=None, extra_right=""): left = f'{ICON_SVG}

Forge Paste

' if slug: left += f'/{slug}' return f"""
{left}
""" def page_footer(host): return f'
Forge Paste v{VERSION} · curl -X POST -d "text" {html_mod.escape(host)}/api/paste
' def page_editor(host, fork_paste=None): stats = get_stats() title_val = "" content_val = "" lang_val = "" if fork_paste: title_val = html_mod.escape(fork_paste.get("title") or "") content_val = html_mod.escape(fork_paste["content"]) lang_val = fork_paste.get("language") or "" lang_options = '' for l in LANGUAGES: sel = ' selected' if l == lang_val else '' lang_options += f'' expiry_options = "" for label, val in [("10 minutes","10m"),("1 hour","1h"),("1 day","1d"),("7 days","7d"),("30 days","30d"),("90 days","90d"),("1 year","1y"),("Never","never")]: sel = ' selected' if val == "30d" else '' expiry_options += f'' return f"""{page_head("Forge Paste")} {page_nav(extra_right=f'{stats["active_pastes"]:,} pastes · {format_bytes(stats["total_size"])}·')}
{page_footer(host)} {TOAST_JS} """ def page_viewer(paste, host, is_new=False, edit_token=None): slug = paste["slug"] lang = paste.get("language") or "plaintext" title = paste.get("title") or slug ta = time_ago(paste["created_at"]) v = paste["views"] size = format_bytes(paste.get("size", len(paste["content"]))) burn = paste.get("burn_after_read", 0) burned = paste.get("_burned", False) parent = paste.get("parent_slug") url = f"{host}/{slug}" lines = paste["content"].split("\n") lc = len(lines) gw = len(str(lc)) expiry_html = "" ea = paste.get("expires_at") if ea: rem = ea - time.time() if rem > 86400: expiry_html = f'expires in {int(rem//86400)}d' elif rem > 3600: expiry_html = f'expires in {int(rem//3600)}h' elif rem > 0: expiry_html = 'expires soon' line_nums = "\n".join(f'{i}' for i in range(1, lc + 1)) code_escaped = html_mod.escape(paste["content"]) if is_new: et_param = f"&et={html_mod.escape(edit_token)}" if edit_token else "" et_display = f'
Edit token: {html_mod.escape(edit_token or "")} — save this to edit/delete later
' if edit_token else "" return f"""{page_head(f"{title} — Forge Paste")} {page_nav(slug)}
🔥

Paste created!

View Raw + New
{et_display} {"
🔥 This paste will be destroyed after being read once
" if burn else ""}
{page_footer(host)}{TOAST_JS}""" burn_html = "" if burned: burn_html = '
🔥 This paste has been burned — this is the only time you can view it.
' elif burn: burn_html = '
🔥 Burn after read — this paste will self-destruct after being viewed
' meta_parts = [ta, f"{v} view{'s' if v!=1 else ''}", size, f"{lc} lines"] meta_str = " · ".join(meta_parts) parent_html = f' · forked from {parent}' if parent else "" right_btns = f""" {meta_str}{parent_html}· Raw ⬇ Download 🍴 Fork""" return f"""{page_head(f"{title} — Forge Paste", HLJS_HEAD)} {page_nav(slug, extra_right=right_btns)}
{html_mod.escape(lang)} {expiry_html} {"🔥 burn" if burn and not burned else ""}
{f'

{html_mod.escape(paste.get("title") or "")}

' if paste.get("title") else ""}
{burn_html}
{line_nums}
{code_escaped}
{page_footer(host)}{TOAST_JS} """ def page_notfound(): return f"""{page_head("Not Found — Forge Paste")} {page_nav()}
🔍

Paste not found

This paste may have expired, been burned, or never existed.

+ Create New Paste
""" def page_recent(host, pastes): items = "" for p in pastes: t = html_mod.escape(p.get("title") or "") slug = p["slug"] lang = p.get("language") or "?" views = p.get("views", 0) ta_str = time_ago(p["created_at"]) sz = format_bytes(p.get("size", 0)) burn_tag = ' 🔥' if p.get("burn_after_read") else "" items += f""" {slug} {t or 'untitled'}{burn_tag} {html_mod.escape(lang)}{sz}{views} views{ta_str} """ return f"""{page_head("Recent — Forge Paste")} {page_nav()}

📋 Recent Pastes

{items if items else '

No pastes yet. Create one!

'}
{page_footer(host)}""" def page_api_docs(host): h = html_mod.escape(host) return f"""{page_head("API — Forge Paste")} {page_nav()}

🔧 API Documentation

Forge Paste provides a full REST API with hastebin compatibility. All endpoints accept and return JSON. Base URL: {h}

POST/api/pasteCreate a new paste

Request (JSON)

FieldTypeRequiredDescription
contentstring✅Paste content (max 2MB)
titlestringOptional title (max 120 chars)
languagestringSyntax language (auto-detected if omitted)
expiresstringExpiry: 10m 1h 1d 7d 30d 90d 1y never
burnboolDestroy after first read
unlistedboolHide from recent list

Alternative: Raw Body

Send raw text as the body with any content type other than application/json:

curl -X POST -d "Hello World" {h}/api/paste

Response

{{
  "key": "a1b2c3d4",
  "url": "{h}/a1b2c3d4",
  "raw_url": "{h}/api/raw/a1b2c3d4",
  "edit_token": "xyz...",
  "expires_at": 1716300000.0
}}

Examples

# Simple paste
curl -X POST -d "your text here" {h}/api/paste

# JSON with options
curl -X POST -H "Content-Type: application/json" \\
  -d '{{"content":"print(42)","language":"python","expires":"1h","burn":true}}' \\
  {h}/api/paste

# Pipe a file
cat script.py | curl -X POST -d @- {h}/api/paste

# With title
curl -X POST -H "Content-Type: application/json" \\
  -d '{{"content":"...","title":"My Config"}}' {h}/api/paste
GET/api/paste/<slug>Get paste metadata & content

Response

{{
  "key": "a1b2c3d4",
  "title": "My Paste",
  "content": "...",
  "language": "python",
  "created_at": 1716200000.0,
  "views": 42,
  "size": 1234,
  "expires_at": 1716300000.0
}}
curl {h}/api/paste/a1b2c3d4
GET/api/raw/<slug>Get raw text content

Returns the paste content as text/plain.

curl {h}/api/raw/a1b2c3d4
GET/api/download/<slug>Download as file

Returns the paste with Content-Disposition: attachment header. Optional ?filename=name.ext query param.

curl -OJ {h}/api/download/a1b2c3d4
PUT/api/paste/<slug>Update a paste

Request

FieldTypeRequiredDescription
edit_tokenstring✅Token from create response
contentstring✅New content
titlestringNew title
languagestringNew language
curl -X PUT -H "Content-Type: application/json" \\
  -d '{{"edit_token":"xyz...","content":"updated"}}' \\
  {h}/api/paste/a1b2c3d4
DELETE/api/paste/<slug>Delete a paste

Request

curl -X DELETE -H "Content-Type: application/json" \\
  -d '{{"edit_token":"xyz..."}}' \\
  {h}/api/paste/a1b2c3d4
GET/api/pastes/recentList recent public pastes

Query: ?limit=20&offset=0

curl {h}/api/pastes/recent?limit=10
GET/api/statsServer statistics
curl {h}/api/stats
{{
  "total_pastes": 420,
  "active_pastes": 380,
  "total_views": 12500,
  "total_size": 2048000,
  "version": "{VERSION}"
}}
GET/api/languagesList supported languages
curl {h}/api/languages
GET/api/healthHealth check
curl {h}/api/health
{{"status":"ok","version":"{VERSION}"}}
POST/api/documentsHastebin-compatible create

Send raw text body. Returns {{"key":"..."}}

curl -X POST -d "hello" {h}/api/documents
GET/api/documents/<slug>Hastebin-compatible get
curl {h}/api/documents/a1b2c3d4
{{"key":"a1b2c3d4","data":"..."}}

🖥️ CLI Aliases

Add these to your .bashrc or .zshrc:

# Paste from clipboard
alias fpaste='xclip -selection clipboard -o | curl -s -X POST -d @- {h}/api/paste | jq -r .url'

# Paste a file
fp() {{ curl -s -X POST -d @"$1" {h}/api/paste | jq -r .url; }}

# Paste from stdin
alias fp-='curl -s -X POST -d @- {h}/api/paste | jq -r .url'

# Get a paste
fpget() {{ curl -s {h}/api/raw/"$1"; }}
{page_footer(host)}""" # ── HTTP Handler ──────────────────────────────────────────── class PasteHandler(BaseHTTPRequestHandler): server_version = f"ForgePaste/{VERSION}" def log_message(self, format, *args): pass # Quiet def _json(self, code, data): body = json.dumps(data).encode() self.send_response(code) self.send_header("Content-Type", "application/json; charset=utf-8") self._cors() self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _html(self, code, content): body = content.encode("utf-8") self.send_response(code) self.send_header("Content-Type", "text/html; charset=utf-8") self._cors() self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _text(self, code, content, filename=None): body = content.encode("utf-8") if isinstance(content, str) else content self.send_response(code) if filename: self.send_header("Content-Type", "application/octet-stream") self.send_header("Content-Disposition", f'attachment; filename="{filename}"') else: self.send_header("Content-Type", "text/plain; charset=utf-8") self._cors() self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _cors(self): self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type, Authorization") 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 _ip(self): return self.headers.get("X-Real-IP") or self.client_address[0] def _read_body(self): try: length = int(self.headers.get("Content-Length", 0)) except (ValueError, TypeError): return None, "Invalid Content-Length" if length > MAX_SIZE: return None, f"Content too large (max {format_bytes(MAX_SIZE)})" return self.rfile.read(length), None def do_OPTIONS(self): self.send_response(204) self._cors() self.end_headers() def do_GET(self): parsed = urlparse(self.path) path = parsed.path.rstrip("/") or "/" query = parse_qs(parsed.query) host = self._host() # Static pages if path == "/": return self._html(200, page_editor(host)) if path == "/recent": pastes = get_recent(limit=50) return self._html(200, page_recent(host, pastes)) if path == "/api" or path == "/api/docs": return self._html(200, page_api_docs(host)) # API: health if path == "/api/health": return self._json(200, {"status": "ok", "version": VERSION}) # API: stats if path == "/api/stats": return self._json(200, get_stats()) # API: languages if path == "/api/languages": return self._json(200, {"languages": LANGUAGES}) # API: recent if path == "/api/pastes/recent": limit = min(int(query.get("limit", [20])[0]), 100) offset = max(int(query.get("offset", [0])[0]), 0) return self._json(200, {"pastes": get_recent(limit, offset)}) # API: raw if path.startswith("/api/raw/"): slug = path[9:] paste = get_paste(slug) if not paste: return self._text(404, "Not found") return self._text(200, paste["content"]) # API: download if path.startswith("/api/download/"): slug = path[14:] paste = get_paste(slug) if not paste: return self._text(404, "Not found") lang = paste.get("language") or "txt" ext_map = {"python":"py","javascript":"js","typescript":"ts","bash":"sh","ruby":"rb", "rust":"rs","csharp":"cs","plaintext":"txt","markdown":"md","yaml":"yml"} ext = ext_map.get(lang, lang) fname = query.get("filename", [f"{slug}.{ext}"])[0] return self._text(200, paste["content"], filename=fname) # API: paste detail if path.startswith("/api/paste/"): slug = path[11:] paste = get_paste(slug) if not paste: return self._json(404, {"error": "not found"}) return self._json(200, { "key": paste["slug"], "title": paste.get("title"), "content": paste["content"], "language": paste.get("language"), "created_at": paste["created_at"], "updated_at": paste.get("updated_at"), "views": paste["views"], "size": paste.get("size"), "expires_at": paste.get("expires_at"), "burn_after_read": bool(paste.get("burn_after_read")), }) # API: hastebin compat if path.startswith("/api/documents/"): slug = path[15:] paste = get_paste(slug) if not paste: return self._json(404, {"error": "not found"}) return self._json(200, {"key": paste["slug"], "data": paste["content"]}) # Web: fork if path.endswith("/fork"): slug = path[1:-5] paste = get_paste(slug, increment_view=False) if not paste: return self._html(404, page_notfound()) paste["parent_slug"] = slug return self._html(200, page_editor(host, fork_paste=paste)) # Web: view paste slug = path[1:] if slug and re.match(r'^[a-z0-9]+$', slug): paste = get_paste(slug) if paste: is_new = "new" in query et = query.get("et", [None])[0] return self._html(200, page_viewer(paste, host, is_new, edit_token=et)) return self._html(404, page_notfound()) self._html(404, page_notfound()) def do_POST(self): body, err = self._read_body() if err: return self._json(400, {"error": err}) parsed = urlparse(self.path) path = parsed.path.rstrip("/") host = self._host() ip = self._ip() # Hastebin compat if path == "/api/documents": text = body.decode("utf-8", errors="replace") if not text.strip(): return self._json(400, {"error": "Empty content"}) slug, _ = create_paste(text, ip=ip) return self._json(200, {"key": slug}) # Main create if path == "/api/paste": ct = self.headers.get("Content-Type", "") if "application/json" in ct: try: data = json.loads(body) except json.JSONDecodeError: return self._json(400, {"error": "Invalid JSON"}) content = data.get("content", "") title = data.get("title") language = data.get("language") expires_key = data.get("expires", "30d") burn = data.get("burn", False) unlisted = data.get("unlisted", False) parent = data.get("parent_slug") else: content = body.decode("utf-8", errors="replace") title = None language = None expires_key = "30d" burn = False unlisted = False parent = None if not content or not content.strip(): return self._json(400, {"error": "Empty content"}) if len(content.encode()) > MAX_SIZE: return self._json(413, {"error": f"Content too large (max {format_bytes(MAX_SIZE)})"}) expires_in = EXPIRY_PRESETS.get(expires_key, EXPIRY_PRESETS["30d"]) slug, edit_token = create_paste( content, title=title, language=language, ip=ip, expires_in=expires_in, burn=burn, is_public=not unlisted, parent_slug=parent ) result = { "key": slug, "url": f"{host}/{slug}", "raw_url": f"{host}/api/raw/{slug}", "edit_token": edit_token, } if expires_in > 0: result["expires_at"] = time.time() + expires_in return self._json(200, result) self._json(404, {"error": "Not found"}) def do_PUT(self): body, err = self._read_body() if err: return self._json(400, {"error": err}) parsed = urlparse(self.path) path = parsed.path.rstrip("/") if path.startswith("/api/paste/"): slug = path[11:] try: data = json.loads(body) except json.JSONDecodeError: return self._json(400, {"error": "Invalid JSON"}) token = data.get("edit_token", "") if not token: return self._json(401, {"error": "edit_token required"}) paste = get_paste(slug, increment_view=False) if not paste: return self._json(404, {"error": "not found"}) if paste.get("edit_token") != token: return self._json(403, {"error": "Invalid edit_token"}) content = data.get("content") if not content or not content.strip(): return self._json(400, {"error": "Empty content"}) update_paste(slug, content, title=data.get("title"), language=data.get("language")) return self._json(200, {"key": slug, "updated": True}) self._json(404, {"error": "Not found"}) def do_DELETE(self): body, err = self._read_body() if err: return self._json(400, {"error": err}) parsed = urlparse(self.path) path = parsed.path.rstrip("/") if path.startswith("/api/paste/"): slug = path[11:] try: data = json.loads(body) if body else {} except json.JSONDecodeError: return self._json(400, {"error": "Invalid JSON"}) token = data.get("edit_token", "") if not token: return self._json(401, {"error": "edit_token required"}) paste = get_paste(slug, increment_view=False) if not paste: return self._json(404, {"error": "not found"}) if paste.get("edit_token") != token: return self._json(403, {"error": "Invalid edit_token"}) delete_paste(slug) return self._json(200, {"key": slug, "deleted": True}) self._json(404, {"error": "Not found"}) # ── Threaded server ───────────────────────────────────────── from socketserver import ThreadingMixIn class ThreadedHTTPServer(ThreadingMixIn, HTTPServer): daemon_threads = True allow_reuse_address = True 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) parser.add_argument("--host", default=HOST) parser.add_argument("--db", default=DB_PATH) parser.add_argument("--max-age", type=int, 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 = ThreadedHTTPServer((HOST, PORT), PasteHandler) print(f"🔥 Forge Paste v{VERSION} running on http://{HOST}:{PORT}") print(f"📦 Database: {DB_PATH}") print(f"⏱ Max age: {MAX_AGE_DAYS} days (0=never)") try: server.serve_forever() except KeyboardInterrupt: print("\nShutting down...") server.server_close() if __name__ == "__main__": main()