"""Inbox Check guest SDK v1.0.0. Python 3.9+, standard library only. Public create/read only, without credentials. Sender email is test input, not registration. Send the tested email yourself using your SMTP/ESP. Never automatically retry a create request after an unknown outcome. """ import http.client import io import json import math import re import socket import time import urllib.parse class InboxCheckError(Exception): def __init__(self, message, status=0, body=None, retry_after=None, timed_out=False): super().__init__(message) self.status = status self.body = body self.retry_after = retry_after self.timed_out = timed_out class InboxCheckTimeout(InboxCheckError): def __init__(self, token, last_result=None): super().__init__('Result wait timed out. Continue reading the same token; do not create another test.', timed_out=True) self.token = token self.last_result = last_result class _DeadlineReader(io.RawIOBase): """Enforce the total budget before each socket read, including headers. An ordinary socket timeout measures inactivity; a trickling peer could otherwise keep response.read() alive indefinitely. """ def __init__(self, connection): super().__init__() self.connection = connection # Preserve socket.makefile's lifetime reference: HTTPConnection may # release its socket immediately after Connection: close headers. self.raw = connection.socket.makefile('rb', buffering=0) def readable(self): return True def readinto(self, buffer): self.connection.set_remaining_timeout() return self.raw.readinto(buffer) def close(self): try: self.raw.close() finally: super().close() class _DeadlineSocket: def __init__(self, socket_value, deadline): self.socket = socket_value self.deadline = deadline def set_remaining_timeout(self): remaining = self.deadline - time.monotonic() if remaining <= 0: raise TimeoutError('Request deadline exceeded') self.socket.settimeout(remaining) def sendall(self, data): self.set_remaining_timeout() return self.socket.sendall(data) def makefile(self, mode, buffering=None): if mode != 'rb': raise ValueError('Only binary HTTP response reading is supported') return io.BufferedReader(_DeadlineReader(self)) def __getattr__(self, name): return getattr(self.socket, name) def _positive(value, name): if isinstance(value, bool) or not isinstance(value, (float, int)) or not math.isfinite(value) or value <= 0: raise ValueError(name + ' must be a positive finite number') return value def _token(value): if not isinstance(value, str) or not re.fullmatch(r'[A-Za-z0-9_-]{1,200}', value): raise ValueError('A valid test token is required') return value def _useful(snapshot): if snapshot.get('status') in ('done', 'expired'): return True rows = snapshot.get('results') return isinstance(rows, list) and any( isinstance(row, dict) and isinstance(row.get('received_at'), str) and bool(row['received_at']) and isinstance(row.get('placement'), str) and row['placement'] not in ('pending', 'not_received', 'unknown', 'error', '') for row in rows ) class InboxCheck: def __init__(self, base_url='https://check.live-direct-marketing.online', request_timeout=30): url = urllib.parse.urlsplit(base_url) if (not url.hostname or url.username or url.password or url.query or url.fragment or url.path not in ('', '/') or (url.scheme != 'https' and not (url.scheme == 'http' and url.hostname in ('localhost', '127.0.0.1', '::1')))): raise ValueError('base_url must be an HTTPS origin (HTTP is allowed only on localhost)') self.base_url = urllib.parse.urlunsplit((url.scheme, url.netloc, '', '', '')) self.request_timeout = _positive(request_timeout, 'request_timeout') def create_test(self, email, **options): """Create once. Options use the API names: providers, emailLocale, senderLinkReports, placementChangeAlertsEnabled, finalReportEnabled. Notification mail is off unless explicitly enabled by the caller. """ if not isinstance(email, str) or len(email) > 200 or not re.fullmatch(r'[^\s@]+@[^\s@]+\.[^\s@]+', email): raise ValueError('A valid sender email is required') allowed = {'providers', 'emailLocale', 'senderLinkReports', 'placementChangeAlertsEnabled', 'finalReportEnabled'} for key in options: if key not in allowed: raise ValueError('Unsupported create option: ' + key) providers = options.get('providers') if 'providers' in options and (not isinstance(providers, list) or len(providers) > 20 or any(not isinstance(p, str) or not p for p in providers)): raise ValueError('providers must be an array of up to 20 provider names') if 'emailLocale' in options and options['emailLocale'] not in ('en', 'ru'): raise ValueError('emailLocale must be en or ru') for key in ('senderLinkReports', 'placementChangeAlertsEnabled', 'finalReportEnabled'): if key in options and not isinstance(options[key], bool): raise ValueError(key + ' must be a boolean') body = {'email': email, 'senderLinkReports': False, 'placementChangeAlertsEnabled': False, 'finalReportEnabled': False} body.update(options) return self._request('POST', '/api/tests', body) def get_test(self, token): return self._request('GET', '/api/tests/' + urllib.parse.quote(_token(token), safe='')) def wait_for_result(self, token, timeout=180, poll_interval=3): """Return first measured or terminal snapshot, including unsent/expired. Inspect results/verdict; return alone does not establish delivery. HTTP errors are surfaced unchanged; polling never creates a new test. """ _token(token) _positive(timeout, 'timeout') _positive(poll_interval, 'poll_interval') deadline = time.monotonic() + timeout last_result = None while True: remaining = deadline - time.monotonic() if remaining <= 0: raise InboxCheckTimeout(token, last_result) try: last_result = self._request('GET', '/api/tests/' + urllib.parse.quote(token, safe=''), timeout=min(self.request_timeout, remaining)) except InboxCheckError as error: if error.timed_out and deadline - time.monotonic() <= 0.005: raise InboxCheckTimeout(token, last_result) from error raise if _useful(last_result): return last_result delay = min(poll_interval, deadline - time.monotonic()) if delay <= 0: raise InboxCheckTimeout(token, last_result) time.sleep(delay) def _request(self, method, path, body=None, timeout=None): headers = {'Accept': 'application/json'} data = None if body is not None: data = json.dumps(body).encode('utf-8') headers['Content-Type'] = 'application/json' url = urllib.parse.urlsplit(self.base_url) budget = timeout if timeout is not None else self.request_timeout deadline = time.monotonic() + budget connection_type = http.client.HTTPSConnection if url.scheme == 'https' else http.client.HTTPConnection connection = connection_type(url.hostname, port=url.port, timeout=budget) try: connection.connect() connection.sock = _DeadlineSocket(connection.sock, deadline) connection.request(method, path, body=data, headers=headers) with connection.getresponse() as response: status = response.status raw = response.read().decode('utf-8', errors='replace') if time.monotonic() >= deadline: raise TimeoutError('Request deadline exceeded') try: parsed = json.loads(raw) if raw else None except ValueError: parsed = {'raw': raw} if status < 200 or status >= 300: raise InboxCheckError('Inbox Check HTTP ' + str(status), status=status, body=parsed, retry_after=response.headers.get('Retry-After')) if not isinstance(parsed, dict) or 'raw' in parsed: raise InboxCheckError('Inbox Check returned an invalid JSON object', status=status, body=parsed) return parsed except InboxCheckError: raise except (OSError, http.client.HTTPException) as error: timed_out = isinstance(error, (socket.timeout, TimeoutError)) or isinstance(getattr(error, 'reason', None), (socket.timeout, TimeoutError)) message = 'Request timed out' if timed_out else 'Network request failed' raise InboxCheckError(message + '; a create request may already have succeeded. Do not automatically repeat it.', timed_out=timed_out) from error finally: connection.close()