Résoudre et soumettre reCAPTCHA avec aiohttp

• Définissez SOLVETHREE_API_KEY sur la clé API de votre espace de travail. Conservez-la sur le serveur.
• Définissez SOLVETHREE_PROXY sur l’URL du proxy que le service doit utiliser pour la résolution. L’envoi à la cible utilise ce même proxy.
• Remplacez accounts.example.com et l’action login par le nom d’hôte et l’action utilisés par votre intégration. Le site doit disposer d’un profil actif accessible à votre espace de travail ; configurez le profil avant de créer des tâches.
• Remplacez l’URL de connexion cible et les données du formulaire par la requête réelle de l’application. L’exemple envoie uniquement g-recaptcha-response ; ajoutez les autres champs requis par votre cible et utilisez le champ de jeton qu’elle attend.

La tâche envoie site, action et un seul proxy. Elle utilise reCAPTCHA Enterprise v3 par défaut ; n’ajoutez pas de champs type ou url. Si le parcours cible nécessite des cookies issus d’une requête antérieure, effectuez ces requêtes via la session cible : l’exemple démarre avec une nouvelle session.

# pip install aiohttp
import asyncio
import os
import uuid
from urllib.parse import urlsplit, unquote
import aiohttp

async def main():
    api = "https://api.solvethree.com/v1"
    proxy = os.environ["SOLVETHREE_PROXY"]
    task = {"site": "accounts.example.com", "action": "login", "proxy": proxy, "requestId": str(uuid.uuid4())}
    task_id = None
    try:
        async with asyncio.timeout(180):
            async with aiohttp.ClientSession(headers={"Authorization": "Bearer " + os.environ["SOLVETHREE_API_KEY"]}) as client:
                async with client.post(api + "/tasks", params={"waitMs": 0}, json=task, allow_redirects=False) as response:
                    response.raise_for_status()
                    if response.status >= 300: raise RuntimeError("Unexpected API redirect")
                    result = await response.json()
                task_id = result["id"]
                while result["status"] in ("pending", "solving"):
                    async with client.get(api + "/tasks/" + task_id, params={"waitMs": 15000}, allow_redirects=False) as response:
                        response.raise_for_status()
                        if response.status >= 300: raise RuntimeError("Unexpected API redirect")
                        result = await response.json()
                    if result["status"] in ("pending", "solving"): await asyncio.sleep(1)
                if result["status"] != "ready":
                    raise RuntimeError("Task failed: " + str(result.get("errorCode", "unknown")))
            solution = result["solution"]
            # A separate session prevents the API Bearer header reaching the target.
            proxy_url = urlsplit(proxy)
            netloc = proxy_url.netloc.rsplit("@", 1)[-1]
            server = f"{proxy_url.scheme}://{netloc}"
            auth = aiohttp.BasicAuth(unquote(proxy_url.username), unquote(proxy_url.password or "")) if proxy_url.username else None
            async with aiohttp.ClientSession() as target:
                async with target.post("https://accounts.example.com/login", proxy=server, proxy_auth=auth,
                    headers={"User-Agent": solution["userAgent"], "sec-ch-ua": solution["secChUa"]},
                    data={"g-recaptcha-response": solution["token"]}, allow_redirects=False) as response:
                    print("Application HTTP status:", response.status)
    except TimeoutError:
        print("Timed out; recover requestId", task["requestId"], "taskId", task_id)
        raise

asyncio.run(main())

Il s’agit d’appels REST asynchrones directs, et non d’une méthode asynchrone du SDK Python officiel. L’exemple effectue un seul POST de création vers /v1/tasks, enregistre l’id renvoyé et utilise des requêtes GET pour interroger cette tâche. pending et solving poursuivent la boucle ; ready autorise l’envoi ; une tâche à l’état failed arrête la boucle.

Le proxy dans le corps de la tâche indique à solvethree quel proxy utiliser. Les arguments proxy et proxy_auth d’aiohttp apparaissent uniquement dans l’appel HTTP vers la cible, pas dans les appels à l’API. L’exemple sépare les identifiants du proxy de l’URL du serveur proxy et envoie le jeton avec les valeurs userAgent et secChUa renvoyées.

Les sessions sont volontairement distinctes : seule la session API possède l’en-tête Authorization. Les redirections sont désactivées et celles de l’API sont traitées comme des erreurs plutôt que suivies.

Une requête à l’API renvoie une erreur HTTP.

L’exemple vérifie le statut HTTP avant de lire le contenu d’une réponse de réussite.

Essayez ceci : Traitez l’erreur HTTP séparément de l’échec de la tâche. Corrigez les entrées non valides ou les problèmes d’autorisation avant de réessayer.

La tâche passe à l’état failed.

Examinez errorCode et error dans la réponse de la tâche. L’exemple lève une exception contenant uniquement errorCode.

Essayez ceci : Arrêtez d’interroger cette tâche et traitez la cause signalée. Corrigez-la avant de réessayer.

La tâche est prête, mais l’application rejette la requête.

Examinez le champ de jeton envoyé, l’URL cible, l’identité du navigateur renvoyée et le proxy, ainsi que les champs de formulaire ou cookies requis.

Essayez ceci : Envoyez solution.token avec l’identité renvoyée via le même proxy. Analysez la réponse de l’application selon les critères de réussite de cette cible : le statut HTTP affiché ne suffit pas à établir l’acceptation.

L’expiration du délai de création ne prouve pas que la création a échoué. L’exemple conserve les entrées d’origine dans task et affiche son requestId ainsi que task_id lorsque le délai expire. Ces variables ne subsistent que tant que le processus s’exécute ; les ID affichés ne constituent pas un enregistrement de reprise complet.

1. 1.

   Enregistrez la requête d’origine avant la création

   Pour permettre une reprise après un redémarrage, conservez de manière sécurisée le corps complet de la tâche, y compris son requestId et son proxy. L’exemple génère un nouvel UUID à chaque exécution ; le relancer tel quel crée donc une nouvelle requête logique.

2. 2.

   Reprenez la tâche existante

   Si vous disposez de task_id, récupérez cette tâche. Si le délai de création a expiré avant la réception d’un ID, répétez la création avec le corps de tâche et le requestId enregistrés, sans les modifier. L’API renvoie la tâche existante ; modifier les entrées tout en réutilisant l’ID provoque un conflit.

3. 3.

   Vérifiez si l’envoi a déjà eu lieu

   Le délai couvre également le POST vers la cible. Si cette requête a expiré, vérifiez le résultat dans l’application avant de l’envoyer à nouveau : celle-ci peut avoir reçu la requête même si votre client n’a pas reçu de réponse. Récupérer la tâche solvethree ne permet pas d’établir ce qui s’est passé côté cible.