vlad/autoreg-skills-max
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
209e142d443042a2d86342c948d06a76c09eb696
autoreg-skills-max/skills/free-tools-abuse/scripts/api-provider-probe.py

188 lines
8.3 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python3
"""
Safe API provider probe for temp-email/proxy testing.
- Does not register accounts with third-party targets.
- Does not store API keys.
- Reads provider keys from environment variables.
- Redacts tokens/proxy passwords/usernames from output.
- Optional Mail.tm lifecycle creates one temporary mailbox and deletes it immediately.
Usage:
python scripts/api-provider-probe.py --mailtm-lifecycle
Optional env vars:
WEBSHARE_TOKEN -> checks Webshare profile + one proxy-list page, redacted
MAILSAC_KEY -> checks Mailsac /api/me, redacted
MAILSLURP_API_KEY -> checks MailSlurp inbox list, redacted
MAILOSAUR_API_KEY -> checks Mailosaur servers endpoint, redacted
"""
from __future__ import annotations
import argparse
import base64
import json
import os
import secrets
import string
import time
import urllib.error
import urllib.request
from typing import Any, Dict, Optional, Tuple
UA = "HermesSafeProbe/1.0"
SENSITIVE_KEYS = {"token", "access_token", "refresh_token", "api_key", "apikey", "key", "password", "proxy_username", "proxy_password", "username"}
def mask(s: str) -> str:
if len(s) <= 8:
return "***"
return s[:4] + "..." + s[-4:]
def redact(obj: Any) -> Any:
if isinstance(obj, dict):
out = {}
for k, v in obj.items():
kl = k.lower()
if kl in SENSITIVE_KEYS or "password" in kl or "token" in kl or "secret" in kl:
out[k] = mask(str(v)) if v is not None else None
elif k in {"results", "data"} and isinstance(v, list):
out[k] = [redact(x) for x in v[:2]]
if len(v) > 2:
out[k].append({"_truncated": len(v) - 2})
else:
out[k] = redact(v)
return out
if isinstance(obj, list):
return [redact(x) for x in obj[:5]] + ([{"_truncated": len(obj) - 5}] if len(obj) > 5 else [])
return obj
def request(method: str, url: str, *, headers: Optional[Dict[str, str]] = None, data: Any = None, timeout: int = 20) -> Tuple[int, Any, float]:
hdr = {"User-Agent": UA, "Accept": "application/json"}
if headers:
hdr.update(headers)
body = None
if data is not None:
body = json.dumps(data).encode("utf-8")
hdr.setdefault("Content-Type", "application/json")
req = urllib.request.Request(url, data=body, headers=hdr, method=method)
t0 = time.time()
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
raw = resp.read().decode("utf-8", "replace")
ms = round((time.time() - t0) * 1000, 1)
try:
parsed = json.loads(raw) if raw else None
except Exception:
parsed = raw[:1000]
return resp.status, parsed, ms
except urllib.error.HTTPError as e:
raw = e.read().decode("utf-8", "replace")
ms = round((time.time() - t0) * 1000, 1)
try:
parsed = json.loads(raw) if raw else raw[:1000]
except Exception:
parsed = raw[:1000]
return e.code, parsed, ms
def domain_from(obj: Any) -> str:
if isinstance(obj, dict):
mem = obj.get("hydra:member") or obj.get("member") or obj.get("domains") or obj.get("data")
if isinstance(mem, list) and mem:
first = mem[0]
return first.get("domain") if isinstance(first, dict) else str(first)
if isinstance(obj, list) and obj:
first = obj[0]
return first.get("domain") if isinstance(first, dict) else str(first)
raise RuntimeError("no domain found")
def probe_mailtm(lifecycle: bool) -> Dict[str, Any]:
base = "https://api.mail.tm"
report: Dict[str, Any] = {"provider": "mail.tm", "auth": "no API key; bearer token after mailbox creation", "steps": []}
st, domains, ms = request("GET", base + "/domains")
report["steps"].append({"step": "domains", "status": st, "ms": ms, "sample": redact(domains)})
if not lifecycle or st != 200:
return report
acct_id = None
token = None
try:
domain = domain_from(domains)
address = "hermes-probe-" + "".join(secrets.choice(string.ascii_lowercase + string.digits) for _ in range(12)) + "@" + domain
password = "P-" + secrets.token_urlsafe(18)
st, acct, ms = request("POST", base + "/accounts", data={"address": address, "password": password})
acct_id = acct.get("id") if isinstance(acct, dict) else None
report["steps"].append({"step": "create_account", "status": st, "ms": ms, "address": address, "account_id": acct_id})
st, tok, ms = request("POST", base + "/token", data={"address": address, "password": password})
token = [REDACTED]"token") if isinstance(tok, dict) else None
report["steps"].append({"step": "get_token", "status": st, "ms": ms, "token": mask(token or ""), "token_len": len(token or "")})
st, me, ms = request("GET", base + "/me", headers={"Authorization": "Bearer " + token})
report["steps"].append({"step": "me", "status": st, "ms": ms, "id_matches": isinstance(me, dict) and me.get("id") == acct_id})
finally:
if acct_id and token:
st, body, ms = request("DELETE", base + f"/accounts/{acct_id}", headers={"Authorization": "Bearer " + token})
report["steps"].append({"step": "delete_account", "status": st, "ms": ms, "deleted": st in (200, 202, 204)})
return report
def probe_webshare() -> Dict[str, Any]:
token = [REDACTED]"WEBSHARE_TOKEN")
report = {"provider": "webshare", "auth": "Authorization: Token <TOKEN>", "configured": bool(token), "steps": []}
if not token:
st, body, ms = request("GET", "https://proxy.webshare.io/api/v2/proxy/list/?mode=direct&page=1&page_size=1")
report["steps"].append({"step": "no_auth_expected_401", "status": st, "ms": ms, "body": body})
return report
hdr = {"Authorization": "Token " + token}
for step, url in [("profile", "https://proxy.webshare.io/api/v2/profile/"), ("proxy_list_one_page", "https://proxy.webshare.io/api/v2/proxy/list/?mode=direct&page=1&page_size=1")]:
st, body, ms = request("GET", url, headers=hdr)
report["steps"].append({"step": step, "status": st, "ms": ms, "body": redact(body)})
return report
def probe_mailsac() -> Dict[str, Any]:
key = os.getenv("MAILSAC_KEY")
report = {"provider": "mailsac", "auth": "Mailsac-Key header", "configured": bool(key), "steps": []}
hdr = {"Mailsac-Key": key} if key else {}
st, body, ms = request("GET", "https://mailsac.com/api/me", headers=hdr)
report["steps"].append({"step": "me", "status": st, "ms": ms, "body": redact(body)})
return report
def probe_mailslurp() -> Dict[str, Any]:
key = os.getenv("MAILSLURP_API_KEY")
report = {"provider": "mailslurp", "auth": "x-api-key", "configured": bool(key), "steps": []}
hdr = {"x-api-key": key} if key else {}
st, body, ms = request("GET", "https://api.mailslurp.com/inboxes?size=1", headers=hdr)
report["steps"].append({"step": "inboxes_page", "status": st, "ms": ms, "body": redact(body)})
return report
def probe_mailosaur() -> Dict[str, Any]:
key = os.getenv("MAILOSAUR_API_KEY")
report = {"provider": "mailosaur", "auth": "HTTP Basic", "configured": bool(key), "steps": []}
if not key:
st, body, ms = request("GET", "https://mailosaur.com/api/servers")
report["steps"].append({"step": "servers_no_auth_expected_401", "status": st, "ms": ms, "body": body})
return report
for label, user, pwd in [("servers_key_as_username", key, ""), ("servers_api_colon_key", "api", key)]:
basic = base64.b64encode(f"{user}:{pwd}".encode()).decode()
st, body, ms = request("GET", "https://mailosaur.com/api/servers", headers={"Authorization": "Basic " + basic})
report["steps"].append({"step": label, "status": st, "ms": ms, "body": redact(body)})
if st < 400:
break
return report
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--mailtm-lifecycle", action="store_true", help="create one Mail.tm mailbox, get token, /me, then delete it")
args = ap.parse_args()
print(json.dumps({"generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "probes": [probe_mailtm(args.mailtm_lifecycle), probe_webshare(), probe_mailsac(), probe_mailslurp(), probe_mailosaur()]}, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink