feat: auto-visit OFTC verification URLs, captcha fallback
Detect /verify/ URLs in NickServ notices and attempt automated verification via SOCKS proxy (POST with token). If the page requires a captcha, save creds as pending and log the URL for manual visit. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -682,6 +682,11 @@ class Network:
|
|||||||
lower = text.lower()
|
lower = text.lower()
|
||||||
log.info("[%s] NickServ: %s", self.cfg.name, text)
|
log.info("[%s] NickServ: %s", self.cfg.name, text)
|
||||||
|
|
||||||
|
# OFTC-style URL verification -- visit the link automatically
|
||||||
|
if "https://" in text and "/verify/" in text:
|
||||||
|
await self._visit_verify_url(text)
|
||||||
|
return
|
||||||
|
|
||||||
if self._nickserv_pending == "identify":
|
if self._nickserv_pending == "identify":
|
||||||
if "you are now identified" in lower:
|
if "you are now identified" in lower:
|
||||||
self._status(f"identified as {self.nick}")
|
self._status(f"identified as {self.nick}")
|
||||||
@@ -752,6 +757,51 @@ class Network:
|
|||||||
log.info("[%s] late NickServ verification confirmation", self.cfg.name)
|
log.info("[%s] late NickServ verification confirmation", self.cfg.name)
|
||||||
await self._on_verify_success()
|
await self._on_verify_success()
|
||||||
|
|
||||||
|
async def _visit_verify_url(self, text: str) -> None:
|
||||||
|
"""Extract and visit a verification URL (e.g. OFTC) via SOCKS proxy.
|
||||||
|
|
||||||
|
Attempts a POST first (OFTC form), falls back to GET.
|
||||||
|
If the page requires a captcha, saves creds as pending and logs the URL.
|
||||||
|
"""
|
||||||
|
import re
|
||||||
|
match = re.search(r'(https?://\S+/verify/\S+)', text)
|
||||||
|
if not match:
|
||||||
|
return
|
||||||
|
url = match.group(1)
|
||||||
|
# Extract token from URL path (after /verify/)
|
||||||
|
token = url.rsplit("/verify/", 1)[-1] if "/verify/" in url else ""
|
||||||
|
log.info("[%s] visiting verification URL: %s", self.cfg.name, url)
|
||||||
|
self._status(f"visiting verification URL...")
|
||||||
|
try:
|
||||||
|
import aiohttp
|
||||||
|
from aiohttp_socks import ProxyConnector
|
||||||
|
connector = ProxyConnector.from_url(
|
||||||
|
f"socks5://{self.proxy_cfg.host}:{self.proxy_cfg.port}",
|
||||||
|
)
|
||||||
|
async with aiohttp.ClientSession(connector=connector) as session:
|
||||||
|
# Try POST with token (OFTC form submission)
|
||||||
|
resp = await session.post(
|
||||||
|
url, data={"token": token},
|
||||||
|
timeout=aiohttp.ClientTimeout(total=15),
|
||||||
|
)
|
||||||
|
body = await resp.text()
|
||||||
|
if resp.status == 200 and "captcha" not in body.lower():
|
||||||
|
log.info("[%s] verification URL accepted (POST %d)",
|
||||||
|
self.cfg.name, resp.status)
|
||||||
|
self._status("verification accepted")
|
||||||
|
await self._on_verify_success()
|
||||||
|
return
|
||||||
|
|
||||||
|
# Captcha or unexpected response -- save as pending, log URL
|
||||||
|
log.warning("[%s] verification requires captcha, visit manually: %s",
|
||||||
|
self.cfg.name, url)
|
||||||
|
self._status(f"verify manually: {url}")
|
||||||
|
await self._on_register_success()
|
||||||
|
except Exception as e:
|
||||||
|
log.warning("[%s] failed to visit verification URL: %s", self.cfg.name, e)
|
||||||
|
self._status(f"verify manually: {url}")
|
||||||
|
await self._on_register_success()
|
||||||
|
|
||||||
def _registration_confirmed(self, lower: str) -> bool:
|
def _registration_confirmed(self, lower: str) -> bool:
|
||||||
"""Check if a NickServ message indicates registration accepted."""
|
"""Check if a NickServ message indicates registration accepted."""
|
||||||
return any(kw in lower for kw in (
|
return any(kw in lower for kw in (
|
||||||
|
|||||||
Reference in New Issue
Block a user