diff --git a/forge-paste.py b/forge-paste.py index 37927e0..6672245 100644 --- a/forge-paste.py +++ b/forge-paste.py @@ -1,42 +1,114 @@ #!/usr/bin/env python3 """ -Forge Paste — A lightweight, self-hosted paste server. -Hastebin-compatible API with a clean web UI. +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 — 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 + 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 -import string -import random -import os -import json -import argparse -import time +import sqlite3, string, random, os, json, argparse, time, hashlib, secrets, re, gzip from http.server import HTTPServer, BaseHTTPRequestHandler -from urllib.parse import urlparse +from urllib.parse import urlparse, parse_qs from datetime import datetime -import html +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 = 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 +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(): - """Get a thread-local database connection.""" db = sqlite3.connect(DB_PATH) db.row_factory = sqlite3.Row db.execute("PRAGMA journal_mode=WAL") @@ -44,37 +116,53 @@ def get_db(): 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, + 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 + 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 ) """) - # Migrate: add expires_at column if missing (existing installs) + # Migrations for 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() - + 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): @@ -82,56 +170,107 @@ def generate_slug(length=8): 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)") + # 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, language=None, ip=None): - cleanup_expired() # Opportunistic cleanup +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() - # Ensure unique while db.execute("SELECT 1 FROM pastes WHERE slug=?", (slug,)).fetchone(): slug = generate_slug() + edit_token = secrets.token_urlsafe(24) now = time.time() - expires_at = now + (MAX_AGE_DAYS * 86400) if MAX_AGE_DAYS > 0 else None + 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, content, language, created_at, views, ip, expires_at) VALUES (?, ?, ?, ?, 0, ?, ?)", - (slug, content, language, now, ip, expires_at) + """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 + return slug, edit_token -def get_paste(slug): +def get_paste(slug, increment_view=True): 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: + 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 dict(row) if row else None + 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() - row = db.execute("SELECT COUNT(*) as total FROM pastes").fetchone() + 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 row["total"] + 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) @@ -140,244 +279,898 @@ def time_ago(ts): if seconds < 86400: return f"{seconds // 3600}h ago" return f"{seconds // 86400}d ago" -# ── HTML Templates ────────────────────────────────────────── +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" -STYLE = """ +# ── HTML Templates ────────────────────────────────────────── +ICON_SVG = '' +FAVICON = "data:image/svg+xml," + +CSS = """ """ -ICON_SVG = '' +HLJS_HEAD = """ + + +""" -def page_editor(host): - total = get_stats() +TOAST_JS = """
+""" + +def page_head(title, extra_head=""): return f""" -Forge Paste{STYLE} -
{ICON_SVG}

Forge Paste

-{total} paste{'s' if total != 1 else ''}
-
Ctrl + Enter to save
-
- -
+{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"])}·')} + +
+ + + +
+ +
+
+ + +
+
+ + + +
+
-
Forge Paste — API: curl -X POST -d "your text" {html.escape(host)}/api/paste
+{page_footer(host)} +{TOAST_JS} """ -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)}
' - +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"] - slug = paste["slug"] + 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}" - 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" - + 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: - return f""" - -{slug} — Forge Paste{STYLE} -
{ICON_SVG}

Forge Paste

-
New
-
✓ Paste created!
-
{html.escape(url)}
-
-View PasteNew Paste
""" - - return f""" - -{slug} — Forge Paste{STYLE} -
{ICON_SVG}

Forge Paste

 -/{slug}
-
{ta} · {v} view{'s' if v != 1 else ''} · {expiry_text} - -Raw -New
-
{line_html}
""" + 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!

+
{html_mod.escape(url)}
+
+ +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""" - -Not Found — Forge Paste{STYLE} -
{ICON_SVG}

Forge Paste

-
🔍

Paste not found

-

This paste may have expired or never existed.

-Create New Paste
""" + 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): - # Quiet logging - pass - - def _send(self, code, content, content_type="text/html"): + pass # Quiet + + def _json(self, code, data): + body = json.dumps(data).encode() 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.send_header("Content-Type", "application/json; charset=utf-8") + self._cors() + self.send_header("Content-Length", str(len(body))) self.end_headers() - if isinstance(content, str): - content = content.encode("utf-8") - self.wfile.write(content) - + 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(204, "") - + self.send_response(204) + self._cors() + self.end_headers() + def do_GET(self): parsed = urlparse(self.path) - path = parsed.path.rstrip("/") - query = parsed.query + path = parsed.path.rstrip("/") or "/" + query = parse_qs(parsed.query) host = self._host() - - if path == "" or path == "/": - return self._send(200, page_editor(host)) - - # API routes + + # 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._send(404, "Not found", "text/plain") - return self._send(200, paste["content"], "text/plain; charset=utf-8") - + 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._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): + 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=1" in query - return self._send(200, page_viewer(paste, host, is_new)) - return self._send(404, page_notfound()) - - self._send(404, page_notfound()) - + 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): - 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") - + 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.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") + 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, 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})") + 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 = HTTPServer((HOST, PORT), PasteHandler) - print(f"🔥 Forge Paste running on http://{HOST}:{PORT}") + + 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: