| """Real AWS CLI through an HTTPS ingress and a trusted HTTP sandbox backend.""" | |
| import asyncio | |
| import json | |
| import os | |
| import ssl | |
| from pathlib import Path | |
| from urllib.parse import urlsplit | |
| import aiohttp | |
| from aiohttp import web | |
| import uvicorn | |
| from yarl import URL | |
| from buckets_aws_cli_e2e import main as aws_cli_checks | |
| ROOT = Path(__file__).resolve().parents[1] | |
| WORK = ROOT / "work/buckets-sandbox" | |
| HOP_HEADERS = { | |
| "connection", | |
| "keep-alive", | |
| "proxy-authenticate", | |
| "proxy-authorization", | |
| "te", | |
| "trailer", | |
| "transfer-encoding", | |
| "upgrade", | |
| } | |
| def end_to_end_headers(headers): | |
| excluded = HOP_HEADERS | {part.strip().lower() for part in headers.get("connection", "").split(",")} | |
| return {name: value for name, value in headers.items() if name.lower() not in excluded} | |
| def browser_post_checks(): | |
| import boto3 | |
| from botocore.config import Config | |
| import httpx | |
| from uuid import uuid4 | |
| from defusedxml.ElementTree import fromstring | |
| ca = ssl.create_default_context(cafile=str(WORK / "localhost.pem")) | |
| tokens = json.loads((WORK / "credentials.json").read_text()) | |
| with httpx.Client( | |
| base_url="https://localhost:28433", | |
| verify=ca, | |
| trust_env=False, | |
| headers={"Authorization": "Bearer " + tokens["user1_org"]}, | |
| ) as native: | |
| result = native.post("/v1/buckets", json={"name": "proxy-post-" + uuid4().hex[:12], "quota_bytes": 1024**2}) | |
| result.raise_for_status() | |
| bucket = result.json() | |
| credential = None | |
| try: | |
| result = native.post( | |
| "/v1/buckets/s3-credentials", | |
| json={ | |
| "permissions": ["storage.read", "storage.write", "storage.admin"], | |
| "bucket_ids": [bucket["id"]], | |
| }, | |
| ) | |
| result.raise_for_status() | |
| credential = result.json() | |
| client = boto3.client( | |
| "s3", | |
| endpoint_url=credential["endpoint"], | |
| region_name="us-east-1", | |
| aws_access_key_id=credential["access_key_id"], | |
| aws_secret_access_key=credential["secret_access_key"], | |
| verify=str(WORK / "localhost.pem"), | |
| config=Config(signature_version="s3v4", s3={"addressing_style": "path"}), | |
| ) | |
| client.put_bucket_policy( | |
| Bucket=bucket["name"], | |
| Policy=json.dumps( | |
| { | |
| "Statement": [ | |
| { | |
| "Effect": "Deny", | |
| "Principal": "*", | |
| "Action": "s3:*", | |
| "Resource": [f"arn:aws:s3:::{bucket['name']}", f"arn:aws:s3:::{bucket['name']}/*"], | |
| "Condition": {"Bool": {"aws:SecureTransport": "false"}}, | |
| } | |
| ] | |
| } | |
| ), | |
| ) | |
| form = client.generate_presigned_post( | |
| Bucket=bucket["name"], | |
| Key="browser.txt", | |
| Fields={"success_action_status": "201"}, | |
| Conditions=[{"success_action_status": "201"}, ["content-length-range", 1, 1024]], | |
| ExpiresIn=60, | |
| ) | |
| with httpx.Client(verify=ca, trust_env=False) as browser: | |
| result = browser.post(form["url"], data=form["fields"], files={"file": ("browser.txt", b"edge TLS")}) | |
| assert result.status_code == 201 | |
| location = fromstring(result.content).find("{*}Location").text | |
| assert location.startswith("https://localhost:28433/") | |
| assert client.get_object(Bucket=bucket["name"], Key="browser.txt")["Body"].read() == b"edge TLS" | |
| finally: | |
| native.delete("/v1/buckets/" + bucket["id"] + "/object", params={"key": "browser.txt"}) | |
| native.delete("/v1/buckets/" + bucket["id"]) | |
| if credential: | |
| native.delete("/v1/buckets/s3-credentials/" + credential["id"]) | |
| print("PASS TLS ingress: org browser POST, public HTTPS Location and SecureTransport policy") | |
| async def main(edge_tls=False): | |
| config = json.loads((WORK / "config.json").read_text()) | |
| database = urlsplit(config["BUCKETS_DATABASE_URL"]) | |
| blob_settings = dict( | |
| part.split("=", 1) for part in config.get("BUCKETS_AZURITE_CONNECTION_STRING", "").split(";") if "=" in part | |
| ) | |
| blob = urlsplit(blob_settings.get("BlobEndpoint", "")) | |
| if ( | |
| config.get("BUCKETS_SANDBOX") != "1" | |
| or database.hostname not in {"localhost", "127.0.0.1"} | |
| or database.port != 55432 | |
| or not database.path.startswith("/buckets") | |
| or config.get("BUCKETS_AZURE_ACCOUNT_URL") | |
| or blob.hostname not in {"localhost", "127.0.0.1"} | |
| or blob.port != 51000 | |
| ): | |
| raise RuntimeError("Proxy drill requires the isolated loopback sandbox") | |
| config["BUCKETS_TLS_TERMINATED"] = "1" if edge_tls else "0" | |
| config["BUCKETS_PUBLIC_ENDPOINT"] = "https://localhost:28433/s3/" | |
| os.environ.update(config) | |
| backend = uvicorn.Server( | |
| uvicorn.Config( | |
| "buckets.sandbox:create_app", | |
| factory=True, | |
| host="127.0.0.1", | |
| port=28432, | |
| proxy_headers=not edge_tls, | |
| forwarded_allow_ips="127.0.0.1", | |
| access_log=False, | |
| log_level="warning", | |
| ) | |
| ) | |
| task = asyncio.create_task(backend.serve()) | |
| runner = None | |
| previous_endpoint = os.environ.get("BUCKETS_E2E_ENDPOINT") | |
| try: | |
| for _ in range(200): | |
| if backend.started: | |
| break | |
| if task.done(): | |
| await task | |
| raise RuntimeError("HTTP backend exited before readiness") | |
| await asyncio.sleep(0.05) | |
| else: | |
| raise RuntimeError("HTTP backend did not start") | |
| async with aiohttp.ClientSession(auto_decompress=False, trust_env=False) as client: | |
| async def proxy(request): | |
| headers = { | |
| name: value | |
| for name, value in end_to_end_headers(request.headers).items() | |
| if name.lower() not in {"x-forwarded-proto", "x-forwarded-for", "forwarded"} | |
| } | |
| # This ingress overwrites caller-supplied forwarding metadata. | |
| # It derives HTTPS from the actual TLS socket, never a header. | |
| if not edge_tls: | |
| headers["X-Forwarded-Proto"] = "https" if request.secure else "http" | |
| headers["X-Forwarded-For"] = request.remote or "" | |
| target = URL("http://127.0.0.1:28432" + request.raw_path, encoded=True) | |
| async with client.request( | |
| request.method, target, headers=headers, data=request.content.iter_any(), allow_redirects=False | |
| ) as upstream: | |
| response = web.StreamResponse(status=upstream.status, headers=end_to_end_headers(upstream.headers)) | |
| await response.prepare(request) | |
| async for chunk in upstream.content.iter_any(): | |
| await response.write(chunk) | |
| await response.write_eof() | |
| return response | |
| app = web.Application(client_max_size=0) | |
| app.router.add_route("*", "/{path:.*}", proxy) | |
| runner = web.AppRunner(app, access_log=None) | |
| await runner.setup() | |
| tls = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) | |
| tls.load_cert_chain(WORK / "localhost.pem", WORK / "localhost-key.pem") | |
| await web.TCPSite(runner, "127.0.0.1", 28433, ssl_context=tls).start() | |
| os.environ["BUCKETS_E2E_ENDPOINT"] = "https://localhost:28433/s3/" | |
| await asyncio.to_thread(aws_cli_checks) | |
| await asyncio.to_thread(browser_post_checks) | |
| await runner.cleanup() | |
| runner = None | |
| print("PASS TLS ingress: real AWS CLI trailers, hashed bodies, presigned GET and multipart across HTTP backend") | |
| finally: | |
| if previous_endpoint is None: | |
| os.environ.pop("BUCKETS_E2E_ENDPOINT", None) | |
| else: | |
| os.environ["BUCKETS_E2E_ENDPOINT"] = previous_endpoint | |
| if runner is not None: | |
| await runner.cleanup() | |
| backend.should_exit = True | |
| await asyncio.wait_for(task, 20) | |
| if __name__ == "__main__": | |
| import argparse | |
| parser = argparse.ArgumentParser() | |
| parser.add_argument("--edge-tls", action="store_true", help="HTTP backend with no trusted forwarding headers") | |
| asyncio.run(main(parser.parse_args().edge_tls)) | |
Xet Storage Details
- Size:
- 8.72 kB
- Xet hash:
- 91c7f10b5e8cce642e8201dbb0ab6586ab96582b1669dc07dd94f75b658d1598
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.