530f33be76
Add 7 new pure-stdlib plugins: whois (raw TCP port 43), portcheck (async TCP connect scan with internal-net guard), httpcheck (HTTP status/redirects/timing), tlscheck (TLS version/cipher/cert inspect), blacklist (parallel DNSBL check against 10 RBLs), rand (password/hex/ uuid/bytes/int/coin/dice), and timer (async countdown notifications). Add --cprofile flag to CLI for profiling bot runtime. Update all docs. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
110 lines
3.3 KiB
Python
110 lines
3.3 KiB
Python
"""Plugin: HTTP status checker (pure stdlib, urllib)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import ssl
|
|
import time
|
|
import urllib.request
|
|
|
|
from derp.plugin import command
|
|
|
|
_TIMEOUT = 10
|
|
_MAX_REDIRECTS = 10
|
|
_USER_AGENT = "derp/1.0"
|
|
|
|
|
|
def _check(url: str) -> dict:
|
|
"""Blocking HTTP check. Returns dict with status info."""
|
|
result: dict = {
|
|
"url": url,
|
|
"status": 0,
|
|
"reason": "",
|
|
"time_ms": 0,
|
|
"redirects": [],
|
|
"server": "",
|
|
"content_type": "",
|
|
"error": "",
|
|
}
|
|
|
|
# Build opener that doesn't follow redirects automatically
|
|
class NoRedirect(urllib.request.HTTPRedirectHandler):
|
|
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
|
result["redirects"].append((code, newurl))
|
|
if len(result["redirects"]) >= _MAX_REDIRECTS:
|
|
return None
|
|
return urllib.request.HTTPRedirectHandler.redirect_request(
|
|
self, req, fp, code, msg, headers, newurl,
|
|
)
|
|
|
|
ctx = ssl.create_default_context()
|
|
ctx.check_hostname = False
|
|
ctx.verify_mode = ssl.CERT_NONE
|
|
|
|
opener = urllib.request.build_opener(
|
|
NoRedirect,
|
|
urllib.request.HTTPSHandler(context=ctx),
|
|
)
|
|
|
|
req = urllib.request.Request(url, method="HEAD")
|
|
req.add_header("User-Agent", _USER_AGENT)
|
|
|
|
t0 = time.monotonic()
|
|
try:
|
|
resp = opener.open(req, timeout=_TIMEOUT)
|
|
result["time_ms"] = (time.monotonic() - t0) * 1000
|
|
result["status"] = resp.status
|
|
result["reason"] = resp.reason
|
|
result["server"] = resp.headers.get("Server", "")
|
|
result["content_type"] = resp.headers.get("Content-Type", "")
|
|
resp.close()
|
|
except urllib.error.HTTPError as exc:
|
|
result["time_ms"] = (time.monotonic() - t0) * 1000
|
|
result["status"] = exc.code
|
|
result["reason"] = exc.reason
|
|
except urllib.error.URLError as exc:
|
|
result["time_ms"] = (time.monotonic() - t0) * 1000
|
|
result["error"] = str(exc.reason)
|
|
except Exception as exc:
|
|
result["time_ms"] = (time.monotonic() - t0) * 1000
|
|
result["error"] = str(exc)
|
|
|
|
return result
|
|
|
|
|
|
@command("httpcheck", help="HTTP check: !httpcheck <url>")
|
|
async def cmd_httpcheck(bot, message):
|
|
"""Check HTTP status, redirects, and response time.
|
|
|
|
Usage:
|
|
!httpcheck https://example.com
|
|
!httpcheck http://10.0.0.1:8080
|
|
"""
|
|
parts = message.text.split(None, 2)
|
|
if len(parts) < 2:
|
|
await bot.reply(message, "Usage: !httpcheck <url>")
|
|
return
|
|
|
|
url = parts[1]
|
|
if not url.startswith(("http://", "https://")):
|
|
url = f"https://{url}"
|
|
|
|
loop = asyncio.get_running_loop()
|
|
result = await loop.run_in_executor(None, _check, url)
|
|
|
|
if result["error"]:
|
|
await bot.reply(message, f"{url} -> error: {result['error']} ({result['time_ms']:.0f}ms)")
|
|
return
|
|
|
|
parts_out = [f"{result['status']} {result['reason']}"]
|
|
parts_out.append(f"{result['time_ms']:.0f}ms")
|
|
|
|
if result["redirects"]:
|
|
chain = " -> ".join(f"{code} {loc}" for code, loc in result["redirects"])
|
|
parts_out.append(f"redirects: {chain}")
|
|
|
|
if result["server"]:
|
|
parts_out.append(f"server: {result['server']}")
|
|
|
|
await bot.reply(message, f"{url} -> {' | '.join(parts_out)}")
|