diff --git a/.gitignore b/.gitignore index 90c5faf..7c353ee 100644 --- a/.gitignore +++ b/.gitignore @@ -13,3 +13,4 @@ build/ .mypy_cache/ config/derp.toml data/ +secrets/ diff --git a/docs/CHEATSHEET.md b/docs/CHEATSHEET.md index db8885c..b880df0 100644 --- a/docs/CHEATSHEET.md +++ b/docs/CHEATSHEET.md @@ -273,6 +273,16 @@ API keys: set `ABUSEIPDB_API_KEY` / `VIRUSTOTAL_API_KEY` env vars or configure in `config/derp.toml` under `[abuseipdb]` / `[virustotal]`. VT rate limit: 4 req/min. Email check: max 5, admin only. +## FlaskPaste + +``` +!paste some text here # Create paste, get URL +!shorten https://long-url.com/x # Shorten URL +``` + +Auto-shortens URLs in `!alert` announcements and history when plugin is loaded. +Config: `[flaskpaste]` in `config/derp.toml` or `FLASKPASTE_URL` env var. + ### Data Setup ```bash diff --git a/docs/USAGE.md b/docs/USAGE.md index 4d732d1..6dfb2e5 100644 --- a/docs/USAGE.md +++ b/docs/USAGE.md @@ -131,6 +131,8 @@ format = "text" # Log format: "text" (default) or "json" | `!abuse report ` | Report IP to AbuseIPDB (admin) | | `!vt ` | VirusTotal lookup | | `!emailcheck [email2 ...]` | SMTP email verification (admin) | +| `!paste ` | Create a paste on FlaskPaste | +| `!shorten ` | Shorten a URL via FlaskPaste | ### Command Shorthand @@ -845,3 +847,49 @@ bad@example.com -- SMTP 550 User unknown (mx: mail.example.com) - Raw SMTP via SOCKS5 proxy: EHLO, MAIL FROM:<>, RCPT TO, QUIT - 15-second timeout per connection - Max 5 emails per invocation + +### `!paste` -- Create Paste + +Upload text to FlaskPaste and get a shareable URL. + +``` +!paste some long text here +``` + +Output format: + +``` +https://paste.mymx.me/abc12345 +``` + +- PoW challenge (difficulty 20) solved per request +- Content sent as JSON body to FlaskPaste API +- No API key needed -- PoW is the auth mechanism +- Raw content available at `/raw` + +### `!shorten` -- Shorten URL + +Shorten a URL via FlaskPaste's URL shortener. + +``` +!shorten https://very-long-url.example.com/path/to/resource +``` + +Output format: + +``` +https://paste.mymx.me/s/AbCdEfGh +``` + +- URL must start with `http://` or `https://` +- PoW challenge (difficulty 20) solved per request +- Also used internally by `!alert` to shorten announcement URLs + +### FlaskPaste Configuration + +```toml +[flaskpaste] +url = "https://paste.mymx.me" # or set FLASKPASTE_URL env var +``` + +TLS: custom CA cert at `secrets/flaskpaste/derp.crt` loaded automatically. diff --git a/plugins/alert.py b/plugins/alert.py index 494e48d..c606096 100644 --- a/plugins/alert.py +++ b/plugins/alert.py @@ -1753,10 +1753,18 @@ async def _poll_once(bot, key: str, announce: bool = True) -> None: if announce and matched: channel = data["channel"] name = data["name"] + fp = bot.registry._modules.get("flaskpaste") for item in matched: short_id = _save_result(channel, name, tag, item) title = _truncate(item["title"]) if item["title"] else "(no title)" url = item["url"] + if fp and url: + try: + url = await loop.run_in_executor( + None, fp.shorten_url, bot, url, + ) + except Exception: + pass date = item.get("date", "") line = f"[{name}/{tag}/{short_id}]" if date: @@ -1936,9 +1944,18 @@ async def cmd_alert(bot, message): if not rows: await bot.reply(message, f"{name}: no history yet") return + loop = asyncio.get_running_loop() + fp = bot.registry._modules.get("flaskpaste") for backend, title, url, date, found_at, short_id in reversed(rows): ts = found_at[:10] title = _truncate(title) if title else "(no title)" + if fp and url: + try: + url = await loop.run_in_executor( + None, fp.shorten_url, bot, url, + ) + except Exception: + pass line = f"[{name}/{backend}/{short_id}] ({date or ts}) {title}" if url: line += f" -- {url}" diff --git a/plugins/flaskpaste.py b/plugins/flaskpaste.py new file mode 100644 index 0000000..bd1e313 --- /dev/null +++ b/plugins/flaskpaste.py @@ -0,0 +1,218 @@ +"""Plugin: FlaskPaste integration -- paste creation and URL shortening.""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +import logging +import os +import ssl +import urllib.request +from pathlib import Path + +from derp.plugin import command + +log = logging.getLogger(__name__) + +_DEFAULT_URL = "https://paste.mymx.me" +_TIMEOUT = 15 +_CERT_DIR = Path(__file__).resolve().parent.parent / "secrets" / "flaskpaste" + + +# -- Internal helpers -------------------------------------------------------- + +def _get_base_url(bot) -> str: + """Resolve FlaskPaste base URL from env or config, strip trailing slash.""" + url = (os.environ.get("FLASKPASTE_URL", "") + or bot.config.get("flaskpaste", {}).get("url", _DEFAULT_URL)) + return url.rstrip("/") + + +def _ssl_context() -> ssl.SSLContext: + """Build SSL context with custom CA cert if available.""" + cert_path = _CERT_DIR / "derp.crt" + if cert_path.exists(): + ctx = ssl.create_default_context(cafile=str(cert_path)) + else: + ctx = ssl.create_default_context() + return ctx + + +def _solve_pow(nonce: str, difficulty: int) -> int: + """Find N where SHA256(nonce:N) has `difficulty` leading zero bits. + + Blocking -- run in executor. + """ + prefix = f"{nonce}:".encode() + n = 0 + while True: + digest = hashlib.sha256(prefix + str(n).encode()).digest() + leading = 0 + for byte in digest: + if byte == 0: + leading += 8 + else: + mask = 0x80 + while mask and not (byte & mask): + leading += 1 + mask >>= 1 + break + if leading >= difficulty: + return n + n += 1 + + +def _get_challenge(base_url: str) -> dict: + """GET /challenge -- returns {nonce, difficulty, token}.""" + ctx = _ssl_context() + req = urllib.request.Request( + f"{base_url}/challenge", + headers={"Accept": "application/json", "User-Agent": "derp-bot"}, + ) + with urllib.request.urlopen(req, timeout=_TIMEOUT, context=ctx) as resp: + return json.loads(resp.read()) + + +def _create_paste(base_url: str, content: str) -> str: + """Challenge + solve + POST / to create a paste. Returns paste URL.""" + ch = _get_challenge(base_url) + nonce = ch["nonce"] + difficulty = ch["difficulty"] + token = ch["token"] + solution = _solve_pow(nonce, difficulty) + + data = json.dumps({"content": content}).encode() + req = urllib.request.Request( + base_url, + data=data, + headers={ + "Content-Type": "application/json", + "X-PoW-Token": token, + "X-PoW-Solution": str(solution), + "User-Agent": "derp-bot", + }, + ) + ctx = _ssl_context() + with urllib.request.urlopen(req, timeout=_TIMEOUT, context=ctx) as resp: + body = json.loads(resp.read()) + # Response should contain the paste URL or ID + paste_id = body.get("id", "") + if paste_id: + return f"{base_url}/{paste_id}" + # Fallback: check for url field + return body.get("url", "") + + +def _shorten_url(base_url: str, url: str) -> str: + """Challenge + solve + POST /s to shorten a URL. Returns short URL.""" + ch = _get_challenge(base_url) + nonce = ch["nonce"] + difficulty = ch["difficulty"] + token = ch["token"] + solution = _solve_pow(nonce, difficulty) + + data = json.dumps({"url": url}).encode() + req = urllib.request.Request( + f"{base_url}/s", + data=data, + headers={ + "Content-Type": "application/json", + "X-PoW-Token": token, + "X-PoW-Solution": str(solution), + "User-Agent": "derp-bot", + }, + ) + ctx = _ssl_context() + with urllib.request.urlopen(req, timeout=_TIMEOUT, context=ctx) as resp: + body = json.loads(resp.read()) + short_id = body.get("id", "") + if short_id: + return f"{base_url}/s/{short_id}" + return body.get("url", "") + + +# -- Public helpers (importable by other plugins) ---------------------------- + +def shorten_url(bot, url: str) -> str: + """Shorten a URL via FlaskPaste. Returns short URL or original on failure.""" + base_url = _get_base_url(bot) + try: + return _shorten_url(base_url, url) + except Exception as exc: + log.debug("flaskpaste shorten failed: %s", exc) + return url + + +def create_paste(bot, content: str) -> str | None: + """Create a paste via FlaskPaste. Returns paste URL or None on failure.""" + base_url = _get_base_url(bot) + try: + return _create_paste(base_url, content) + except Exception as exc: + log.debug("flaskpaste paste failed: %s", exc) + return None + + +# -- Commands ---------------------------------------------------------------- + +@command("paste", help="Create a paste: !paste ") +async def cmd_paste(bot, message): + """Create a paste on FlaskPaste and return the URL. + + Usage: + !paste some long text here + """ + parts = message.text.split(None, 1) + if len(parts) < 2: + await bot.reply(message, "Usage: !paste ") + return + + content = parts[1] + base_url = _get_base_url(bot) + loop = asyncio.get_running_loop() + + try: + url = await loop.run_in_executor(None, _create_paste, base_url, content) + except Exception as exc: + await bot.reply(message, f"paste failed: {exc}") + return + + if url: + await bot.reply(message, url) + else: + await bot.reply(message, "paste failed: no URL returned") + + +@command("shorten", help="Shorten a URL: !shorten ") +async def cmd_shorten(bot, message): + """Shorten a URL via FlaskPaste. + + Usage: + !shorten https://very-long-url.example.com/path + """ + parts = message.text.split(None, 1) + if len(parts) < 2: + await bot.reply(message, "Usage: !shorten ") + return + + target_url = parts[1].strip() + if not target_url.startswith(("http://", "https://")): + await bot.reply(message, "URL must start with http:// or https://") + return + + base_url = _get_base_url(bot) + loop = asyncio.get_running_loop() + + try: + short = await loop.run_in_executor( + None, _shorten_url, base_url, target_url, + ) + except Exception as exc: + await bot.reply(message, f"shorten failed: {exc}") + return + + if short: + await bot.reply(message, short) + else: + await bot.reply(message, "shorten failed: no URL returned")