"""Bounded aiohttp HTTP proxy requests; --check performs no network operation.

Set BP_PROXY_HOST, BP_PROXY_PORT, BP_PROXY_USER, BP_PROXY_PASSWORD,
and BP_TARGET_URL (an endpoint you control). Requires aiohttp for network runs.
"""
import argparse
import asyncio
import os
from urllib.parse import quote, urlsplit


def configuration():
    host = os.environ["BP_PROXY_HOST"]
    if not host or any(c in host for c in "/@?#:\\") or any(c.isspace() for c in host):
        raise ValueError("Use an IPv4 address or hostname")
    port = int(os.environ["BP_PROXY_PORT"])
    if not 1 <= port <= 65535:
        raise ValueError("Invalid port")
    user, password = os.environ["BP_PROXY_USER"], os.environ["BP_PROXY_PASSWORD"]
    if ":" in user:
        raise ValueError("Basic authentication usernames cannot contain a colon")
    # Match older aiohttp BasicAuth encoding explicitly across 3.x versions.
    (user + ":" + password).encode("latin1")
    target = os.environ["BP_TARGET_URL"]
    parsed = urlsplit(target)
    if parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.username is not None:
        raise ValueError("Target must be an HTTP(S) URL without userinfo")
    proxy = f"http://{quote(user, safe='')}:{quote(password, safe='')}@{host}:{port}"
    return proxy, target


async def run(proxy, target):
    import aiohttp
    timeout = aiohttp.ClientTimeout(total=30, connect=10, sock_connect=5, sock_read=15)
    connector = aiohttp.TCPConnector(limit=4, limit_per_host=4)
    gate = asyncio.Semaphore(4)
    async with aiohttp.ClientSession(timeout=timeout, connector=connector,
                                    trust_env=False,
                                    cookie_jar=aiohttp.DummyCookieJar()) as session:
        async def fetch(index):
            async with gate:
                try:
                    async with session.get(target, proxy=proxy,
                                           allow_redirects=False) as response:
                        response.raise_for_status()
                        size = 0
                        async for chunk in response.content.iter_chunked(65536):
                            size += len(chunk)
                            if size > 2_000_000:
                                print(f"response_too_large job={index}")
                                return False
                        print(f"request_ok job={index} status={response.status} bytes={size}")
                        return True
                except aiohttp.ClientHttpProxyError as exc:
                    print(f"proxy_error job={index} status={exc.status}")
                except aiohttp.ClientResponseError as exc:
                    print(f"http_error job={index} status={exc.status}")
                except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
                    # Do not serialize the exception, URL, or authentication header.
                    print(f"request_failed job={index} type={type(exc).__name__}")
                return False
        results = await asyncio.gather(*(fetch(i) for i in range(4)))
    # Short-lived HTTPS scripts may need transport cleanup before loop shutdown.
    await asyncio.sleep(0.250)
    return 0 if all(results) else 1


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--check", action="store_true")
    args = parser.parse_args()
    try:
        config = configuration()
    except (KeyError, ValueError, UnicodeError):
        print("configuration_error: check required environment variables and URL format")
        return 2
    if args.check:
        print("configuration_ok scheme=http credentials=redacted network=none")
        return 0
    return asyncio.run(run(*config))


if __name__ == "__main__":
    raise SystemExit(main())