# workflows/temp_mail.py """ Universeller Temp-Mail Service mit Multi-Provider Fallback (mail.tm & 1secmail). Bietet asynchrone Postfächer für automatische Registrierungen und OTP/Link-Parsing. """ import asyncio import logging import random import re import string import httpx from typing import Optional, Tuple logger = logging.getLogger(__name__) MAIL_TM_BASE = "https://api.mail.tm" ONESECMAIL_BASE = "https://www.1secmail.com/api/v1/" def generate_random_username(prefix: str = "agent") -> str: suffix = "".join(random.choices(string.ascii_lowercase + string.digits, k=8)) return f"{prefix}_{suffix}" def generate_random_password(length: int = 16) -> str: chars = string.ascii_letters + string.digits + "!@#$%^&*" return "".join(random.choices(chars, k=length)) class TempMailAccount: def __init__(self, email: str, password: str, provider: str, token: str = "", account_id: str = ""): self.email = email self.password = password self.provider = provider self.token = token self.account_id = account_id self.username, self.domain = email.split("@") if "@" in email else ("", "") class TempMailService: """ Verwaltet temporäre E-Mail-Postfächer mit automatischem Polling nach OTP-Codes und Links. """ def __init__(self, timeout: float = 30.0): self._http = httpx.AsyncClient(timeout=timeout) async def aclose(self): await self._http.aclose() async def create_inbox(self) -> TempMailAccount: """ Erstellt ein temporäres Postfach. Versucht zuerst mail.tm, fällt bei Bedarf auf 1secmail zurück. """ try: return await self._create_mail_tm_inbox() except Exception as e: logger.warning(f"mail.tm fehlgeschlagen ({e}), wechsle zu 1secmail...") return await self._create_1secmail_inbox() async def _create_mail_tm_inbox(self) -> TempMailAccount: # 1. Domains holen domains_resp = await self._http.get(f"{MAIL_TM_BASE}/domains") domains_resp.raise_for_status() domains = domains_resp.json().get("hydra:member", []) if not domains: raise RuntimeError("Keine mail.tm Domains verfügbar") domain = domains[0]["domain"] username = generate_random_username() password = generate_random_password() email = f"{username}@{domain}" # 2. Account registrieren reg_resp = await self._http.post( f"{MAIL_TM_BASE}/accounts", json={"address": email, "password": password}, ) reg_resp.raise_for_status() account_id = reg_resp.json().get("id", "") # 3. JWT Token holen token_resp = await self._http.post( f"{MAIL_TM_BASE}/token", json={"address": email, "password": password}, ) token_resp.raise_for_status() token = token_resp.json().get("token", "") return TempMailAccount( email=email, password=password, provider="mail.tm", token=token, account_id=account_id, ) async def _create_1secmail_inbox(self) -> TempMailAccount: resp = await self._http.get(f"{ONESECMAIL_BASE}?action=genRandomMailbox&count=1") resp.raise_for_status() emails = resp.json() if not emails: raise RuntimeError("Keine 1secmail Postfächer verfügbar") email = emails[0] password = generate_random_password() return TempMailAccount( email=email, password=password, provider="1secmail", ) async def poll_for_otp_code( self, account: TempMailAccount, pattern: str = r"\b\d{4,8}\b", max_wait_seconds: int = 60, poll_interval: float = 3.0, ) -> Optional[str]: """ Wartet auf eine eingehende E-Mail und extrahiert den ersten passenden OTP-Verifizierungscode. """ elapsed = 0.0 while elapsed < max_wait_seconds: messages = await self._fetch_messages(account) for msg in messages: content = await self._fetch_message_content(account, msg) match = re.search(pattern, content) if match: code = match.group(0) logger.info(f"OTP-Code erfolgreich gefunden: {code}") return code await asyncio.sleep(poll_interval) elapsed += poll_interval logger.warning(f"Kein OTP-Code nach {max_wait_seconds}s empfangen.") return None async def poll_for_activation_link( self, account: TempMailAccount, url_pattern: str = r"https?://[^\s\"'>]+(?:verify|confirm|activate|token|auth)[^\s\"'>]*", max_wait_seconds: int = 60, poll_interval: float = 3.0, ) -> Optional[str]: """ Wartet auf eine E-Mail und extrahiert den Aktivierungs-/Bestätigungslink. """ elapsed = 0.0 while elapsed < max_wait_seconds: messages = await self._fetch_messages(account) for msg in messages: content = await self._fetch_message_content(account, msg) match = re.search(url_pattern, content, re.IGNORECASE) if match: link = match.group(0) logger.info(f"Aktivierungslink gefunden: {link}") return link await asyncio.sleep(poll_interval) elapsed += poll_interval logger.warning(f"Kein Aktivierungslink nach {max_wait_seconds}s empfangen.") return None async def _fetch_messages(self, account: TempMailAccount) -> list: try: if account.provider == "mail.tm": headers = {"Authorization": f"Bearer {account.token}"} resp = await self._http.get(f"{MAIL_TM_BASE}/messages", headers=headers) if resp.status_code == 200: return resp.json().get("hydra:member", []) elif account.provider == "1secmail": url = f"{ONESECMAIL_BASE}?action=getMessages&login={account.username}&domain={account.domain}" resp = await self._http.get(url) if resp.status_code == 200: return resp.json() except Exception as e: logger.debug(f"Fehler beim Abrufen der Mails: {e}") return [] async def _fetch_message_content(self, account: TempMailAccount, msg: dict) -> str: try: if account.provider == "mail.tm": msg_id = msg.get("id") headers = {"Authorization": f"Bearer {account.token}"} resp = await self._http.get(f"{MAIL_TM_BASE}/messages/{msg_id}", headers=headers) if resp.status_code == 200: data = resp.json() return data.get("text", "") or data.get("intro", "") or str(data.get("html", "")) elif account.provider == "1secmail": msg_id = msg.get("id") url = f"{ONESECMAIL_BASE}?action=readMessage&login={account.username}&domain={account.domain}&id={msg_id}" resp = await self._http.get(url) if resp.status_code == 200: data = resp.json() return data.get("textBody", "") or data.get("body", "") except Exception as e: logger.debug(f"Fehler beim Lesen des E-Mail-Inhalts: {e}") return ""