#!/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 ", "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()