Auto-backup: 206 scripts (@ 2026-09-17 11:46)

skills/devops/nostr-relay-hosting/scripts/test_nostr_relay.py

@@ -0,0 +1,23 @@
+#!/usr/bin/env python3
+"""Test de connectivité d'un relay Nostr — ws:// local ou wss:// via tunnel.
+Usage: python3 test_nostr_relay.py [url1 url2 ...]   (defaut: ws://127.0.0.1:3001)
+Exige: python3 -c "import websockets" (dispo sur openclaw-1, sinon: pip install websockets)
+Attendu sur relay fermé (auth_required): réponse ["AUTH", "<challenge>"] = connexion OK.
+Piège: le handshake curl ne renvoie PAS 101 (il renvoie 200) — utiliser ce script à la place."""
+import asyncio, json, sys, websockets
+
+async def test(url):
+    try:
+        async with websockets.connect(url, open_timeout=15) as ws:
+            await ws.send(json.dumps(["REQ", "test-conn", {"kinds": [0], "limit": 1}]))
+            msg = await asyncio.wait_for(ws.recv(), timeout=10)
+            return f"OK — réponse: {msg[:120]}"
+    except Exception as e:
+        return f"FAIL — {type(e).__name__}: {str(e)[:150]}"
+
+async def main():
+    urls = sys.argv[1:] or ["ws://127.0.0.1:3001"]
+    for u in urls:
+        print(f"{u} → {await test(u)}")
+
+asyncio.run(main())

skills/software-development/github/scripts/git-credential-token.py

@@ -0,0 +1,65 @@
+#!/usr/bin/env python3
+"""Print the first unambiguous GitHub token in a git credential-store file."""
+
+from pathlib import Path
+import re
+import sys
+from urllib.parse import unquote, urlsplit
+
+
+_TOKEN_PREFIXES = ("ghp_", "github_pat_", "gho_", "ghu_", "ghs_", "ghr_")
+_INVALID_ESCAPE = re.compile(r"%(?![0-9A-Fa-f]{2})")
+
+
+def _decode(value: str | None) -> str:
+    if value is None or _INVALID_ESCAPE.search(value):
+        return ""
+    decoded = unquote(value)
+    if not decoded or any(ord(char) <= 0x1F or 0x7F <= ord(char) <= 0x9F for char in decoded):
+        return ""
+    return decoded
+
+
+def _token_from_url(line: str) -> str:
+    if "\r" in line or "\n" in line:
+        return ""
+    try:
+        credential = urlsplit(line)
+        port = credential.port
+    except ValueError:
+        return ""
+    if credential.scheme != "https" or credential.hostname != "github.com" or port not in (None, 443):
+        return ""
+
+    username = _decode(credential.username)
+    password = _decode(credential.password)
+    if not username:
+        return ""
+    if password and password != "x-oauth-basic":
+        return password
+    if password == "x-oauth-basic":
+        return username
+    return username if username.startswith(_TOKEN_PREFIXES) else ""
+
+
+def main() -> int:
+    path = Path(sys.argv[1]).expanduser() if len(sys.argv) > 1 else Path.home() / ".git-credentials"
+    try:
+        lines = path.read_bytes().split(b"\n")
+    except OSError:
+        return 1
+
+    for raw_line in lines:
+        try:
+            line = raw_line.decode("utf-8")
+        except UnicodeDecodeError:
+            continue
+        token = _token_from_url(line)
+        if token:
+            print(token)
+            return 0
+    return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

skills/web/blocked-page-recovery/scripts/recover_page.py

@@ -0,0 +1,241 @@
+#!/usr/bin/env python3
+"""Recover a blocked / paywalled / WAF'd page from third-party copies.
+
+Ladder (cheapest first):
+  1. Wayback Machine "available" API  -> dated snapshot   (provenance: snapshot)
+  2. archive.today domain rotation    -> dated snapshot   (provenance: snapshot)
+  3. Jina Reader (JINA_API_KEY only)  -> live re-render   (provenance: live)
+
+Every candidate body is validated before being declared a win: byte floors,
+redirect-stub detection (meta-refresh/JS pointing back at the original host),
+and interstitial-title rejection. Fake 200s are the norm in this space.
+
+Stdlib only. Usage:
+    python3 recover_page.py URL [--json] [--out FILE] [--timeout N]
+
+Exit codes: 0 recovered, 1 nothing worked, 2 bad invocation.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import re
+import sys
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+
+USER_AGENT = (
+    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
+    "(KHTML, like Gecko) Chrome/126.0 Safari/537.36"
+)
+
+ARCHIVE_TODAY_HOSTS = ["archive.ph", "archive.md", "archive.li", "archive.is"]
+
+# Titles that mean "this is not the page you asked for".
+INTERSTITIAL_TITLES = (
+    "just a moment",
+    "redirecting",
+    "google search",
+    "attention required",
+    "access denied",
+    "are you a robot",
+    "one more step",
+)
+
+# Below these floors a body is a stub or an error page, not content.
+MIN_BODY_BYTES = {"wayback": 3072, "archive_today": 3072, "jina": 512}
+
+REDIRECT_STUB_RE = re.compile(
+    r'http-equiv=["\']?refresh|window\.location|location\.replace', re.IGNORECASE
+)
+TITLE_RE = re.compile(r"<title[^>]*>(.*?)</title>", re.IGNORECASE | re.DOTALL)
+
+
+def _fetch(
+    url: str,
+    timeout: int,
+    headers: dict | None = None,
+    retries_on_429: int = 2,
+) -> tuple[int, bytes]:
+    req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT, **(headers or {})})
+    for attempt in range(retries_on_429 + 1):
+        try:
+            with urllib.request.urlopen(req, timeout=timeout) as resp:
+                return resp.status, resp.read()
+        except urllib.error.HTTPError as exc:
+            if exc.code == 429 and attempt < retries_on_429:
+                time.sleep(5 * (attempt + 1))
+                continue
+            return exc.code, exc.read() if exc.fp else b""
+        except (urllib.error.URLError, OSError, ValueError):
+            return 0, b""
+    return 0, b""
+
+
+def _fetch_follow(url: str, timeout: int) -> tuple[int, bytes, str]:
+    """Like _fetch but also returns the final URL after redirects."""
+    req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
+    try:
+        with urllib.request.urlopen(req, timeout=timeout) as resp:
+            return resp.status, resp.read(), resp.geturl()
+    except urllib.error.HTTPError as exc:
+        return exc.code, exc.read() if exc.fp else b"", exc.geturl() or url
+    except (urllib.error.URLError, OSError, ValueError):
+        return 0, b"", url
+
+
+def _page_title(body: bytes) -> str:
+    m = TITLE_RE.search(body[:65536].decode("utf-8", "replace"))
+    return re.sub(r"\s+", " ", m.group(1)).strip().lower() if m else ""
+
+
+def validate(body: bytes, route: str, target_url: str) -> str | None:
+    """Return a rejection reason, or None if the body looks like real content."""
+    floor = MIN_BODY_BYTES.get(route, 3072)
+    if len(body) < floor:
+        return f"body_too_small:{len(body)}<{floor}"
+    title = _page_title(body)
+    for marker in INTERSTITIAL_TITLES:
+        if marker in title:
+            return f"interstitial_title:{marker!r}"
+    # Redirect stub: small-ish page whose only job is bouncing back to the
+    # original (blocked) host — the classic AMP-cache failure mode.
+    if len(body) < 8192 and REDIRECT_STUB_RE.search(body.decode("utf-8", "replace")):
+        target_host = urllib.parse.urlsplit(target_url).hostname or ""
+        if target_host and target_host.encode() in body:
+            return "redirect_stub_to_origin"
+    return None
+
+
+def try_wayback(url: str, timeout: int) -> dict | None:
+    snap_url = None
+    snap_ts = None
+    discovery = "https://archive.org/wayback/available?url=" + urllib.parse.quote(url, safe="")
+    status, raw = _fetch(discovery, timeout)
+    if status == 200:
+        try:
+            closest = json.loads(raw).get("archived_snapshots", {}).get("closest", {})
+        except (json.JSONDecodeError, AttributeError):
+            closest = {}
+        if closest.get("available") and closest.get("url"):
+            snap_url = closest["url"].replace(
+                "http://web.archive.org", "https://web.archive.org"
+            )
+            snap_ts = closest.get("timestamp")
+    if snap_url is None:
+        # Discovery API is rate-limited far more aggressively than snapshot
+        # serving. Fall back to the redirect form: /web/2/<url> bounces to
+        # the newest snapshot if one exists (404 page otherwise).
+        snap_url = "https://web.archive.org/web/2/" + url
+    status, body, final_url = _fetch_follow(snap_url, timeout)
+    if status != 200 or validate(body, "wayback", url):
+        return None
+    if snap_ts is None:
+        m = re.search(r"/web/(\d{14})", final_url)
+        snap_ts = m.group(1) if m else None
+    return {
+        "route": "wayback",
+        "provenance": "snapshot",
+        "snapshot_timestamp": snap_ts,
+        "source_url": final_url,
+        "body": body,
+    }
+
+
+def try_archive_today(url: str, timeout: int) -> dict | None:
+    for host in ARCHIVE_TODAY_HOSTS:
+        fetch_url = f"https://{host}/newest/{url}"
+        status, body = _fetch(fetch_url, timeout)
+        if status != 200:
+            continue
+        if validate(body, "archive_today", url):
+            continue  # 429 bodies and interstitials land here
+        return {
+            "route": f"archive_today:{host}",
+            "provenance": "snapshot",
+            "snapshot_timestamp": None,  # archive.today embeds the date in-page
+            "source_url": fetch_url,
+            "body": body,
+        }
+    return None
+
+
+def try_jina(url: str, timeout: int) -> dict | None:
+    key = os.environ.get("JINA_API_KEY")
+    if not key:
+        return None
+    status, body = _fetch(
+        "https://r.jina.ai/" + url, timeout, headers={"Authorization": f"Bearer {key}"}
+    )
+    if status != 200 or validate(body, "jina", url):
+        return None
+    return {
+        "route": "jina_reader",
+        "provenance": "live",
+        "snapshot_timestamp": None,
+        "source_url": "https://r.jina.ai/" + url,
+        "body": body,
+    }
+
+
+ROUTES = (try_wayback, try_archive_today, try_jina)
+
+
+def recover(url: str, timeout: int = 25) -> dict | None:
+    for route_fn in ROUTES:
+        result = route_fn(url, timeout)
+        if result:
+            return result
+    return None
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser(
+        description="Recover a blocked / paywalled / WAF'd page from third-party copies."
+    )
+    ap.add_argument("url")
+    ap.add_argument("--json", action="store_true", help="print metadata as JSON")
+    ap.add_argument("--out", help="write recovered body to this file")
+    ap.add_argument("--timeout", type=int, default=25)
+    args = ap.parse_args()
+
+    if not args.url.startswith(("http://", "https://")):
+        print("error: URL must start with http:// or https://", file=sys.stderr)
+        return 2
+
+    result = recover(args.url, args.timeout)
+    if not result:
+        msg = {"recovered": False, "url": args.url,
+               "hint": "No archive copy found. Try the API-first pivot or the browser tool."}
+        print(json.dumps(msg, indent=2) if args.json else msg["hint"], file=sys.stderr)
+        return 1
+
+    body = result.pop("body")
+    result.update({"recovered": True, "url": args.url, "body_bytes": len(body)})
+    if args.out:
+        with open(args.out, "wb") as fh:
+            fh.write(body)
+        result["saved_to"] = args.out
+
+    if args.json:
+        print(json.dumps(result, indent=2))
+    else:
+        for k, v in result.items():
+            print(f"{k}: {v}")
+        if not args.out:
+            print("\n--- body (first 2000 chars) ---")
+            print(body[:2000].decode("utf-8", "replace"))
+    if result["provenance"] == "snapshot":
+        print(
+            "\nNOTE: this is an ARCHIVED SNAPSHOT, not the live page. "
+            "Cite it with its timestamp.",
+            file=sys.stderr,
+        )
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())