AWS 자원과 코드 매핑

set -euo pipefail
export ROBOT_REPO=/home/ec2-user/my-project/robot-arm
export ROBOT_TEAM=team01
export ROBOT_CONFIG="$ROBOT_REPO/workshop/.local/$ROBOT_TEAM.json"
ROBOT_INFO="$(python3 "$ROBOT_REPO/workshop/scripts/lab.py" info --config "$ROBOT_CONFIG")"
export ROBOT_LAB="$(printf '%s' "$ROBOT_INFO" | python3 -c 'import json,sys; print(json.load(sys.stdin)["workspace"])')"
cd "$ROBOT_LAB"
test -f infra/workshop-binding.json
source "$ROBOT_LAB/.venv/bin/activate"

export ROBOT_RELEASE="$(cat .local/evidence/release.txt)"
export ROBOT_ARTIFACTS="$ROBOT_LAB/infra/.state/releases/$ROBOT_RELEASE.json"
test -f "$ROBOT_ARTIFACTS"

cd "$ROBOT_LAB"
test -f infra/workshop-binding.json
umask 077
mkdir -p .local/tools .local/evidence
(
set -o noclobber
cat > .local/tools/verify-aws.py <<'PY'
"""Participant AWS/HTTPS checks. No model call or new simulation."""
import argparse
from http.cookiejar import CookieJar
import hmac
import json
from pathlib import Path
import sys
import urllib.request

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--skip-http", action="store_true")
    args = parser.parse_args()
    root = Path.cwd()
    if not (root / "infra/workshop-binding.json").is_file():
        raise ValueError("Run from the prepared participant app")
    sys.path.insert(0, str(root / "infra"))
    from deployment import Cloud, PROJECT, STACKS, read_config
    cloud = Cloud(read_config(root / "infra/config.json"))
    cfg, checks = cloud.config, []
    def check(name, passed):
        row = {"check": name, "passed": bool(passed)}
        checks.append(row)
        print(json.dumps(row), flush=True)
        if not passed:
            raise AssertionError(name)
    for key in STACKS:
        if key == "agentcore" and cfg["plannerMode"] != "agentcore":
            continue
        stack = cloud.stack(key)
        check("stack:" + key, stack and stack["StackStatus"] in
              ("CREATE_COMPLETE", "UPDATE_COMPLETE"))
    app, foundation = cloud.outputs("app"), cloud.outputs("foundation")
    service = cloud.client("ecs").describe_services(
        cluster=app["ClusterName"], services=[app["ServiceName"]])["services"][0]
    network = service["networkConfiguration"]["awsvpcConfiguration"]
    check("ecs-count", (service["desiredCount"], service["runningCount"],
                       service["pendingCount"]) == (1, 1, 0))
    check("ecs-private", network["assignPublicIp"] == "DISABLED"
          and set(network["subnets"]) == set(cfg["privateSubnetIds"])
          and network["securityGroups"] == [app["TaskSecurityGroupId"]])
    task = cloud.client("ecs").describe_task_definition(
        taskDefinition=service["taskDefinition"])["taskDefinition"]
    simulation = next(c for c in task["containerDefinitions"] if c["name"] == "simulation")
    check("arm64-private-simulation", task["networkMode"] == "awsvpc"
          and task["runtimePlatform"]["cpuArchitecture"] == "ARM64"
          and simulation.get("essential") is True and not simulation.get("portMappings"))
    alb = cloud.client("elbv2").describe_load_balancers(
        LoadBalancerArns=[app["LoadBalancerArn"]])["LoadBalancers"][0]
    check("public-alb", alb["Scheme"] == "internet-facing"
          and {a["SubnetId"] for a in alb["AvailabilityZones"]} == set(cfg["publicSubnetIds"])
          and alb["SecurityGroups"] == [app["AlbSecurityGroupId"]])
    groups = cloud.client("ec2").describe_security_groups(
        GroupIds=[app["AlbSecurityGroupId"], app["TaskSecurityGroupId"]])["SecurityGroups"]
    groups = {g["GroupId"]: g for g in groups}
    ingress = groups[app["AlbSecurityGroupId"]]["IpPermissions"]
    check("alb-prefix-only", len(ingress) == 1
          and ingress[0].get("FromPort") == 443 and ingress[0].get("ToPort") == 443
          and ingress[0].get("IpProtocol") == "tcp"
          and not ingress[0].get("IpRanges") and not ingress[0].get("Ipv6Ranges")
          and not ingress[0].get("UserIdGroupPairs")
          and [p["PrefixListId"] for p in ingress[0].get("PrefixListIds", [])]
          == [cfg["cloudFrontPrefixListId"]])
    ingress = groups[app["TaskSecurityGroupId"]]["IpPermissions"]
    check("ecs-alb-only", len(ingress) == 1
          and ingress[0].get("FromPort") == 8098 and ingress[0].get("ToPort") == 8098
          and ingress[0].get("IpProtocol") == "tcp"
          and not ingress[0].get("IpRanges") and not ingress[0].get("Ipv6Ranges")
          and not ingress[0].get("PrefixListIds")
          and [g["GroupId"] for g in ingress[0].get("UserIdGroupPairs", [])]
          == [app["AlbSecurityGroupId"]])
    listeners = cloud.client("elbv2").describe_listeners(
        LoadBalancerArn=app["LoadBalancerArn"])["Listeners"]
    check("origin-https", len(listeners) == 1
          and listeners[0]["Protocol"] == "HTTPS" and listeners[0]["Port"] == 443
          and any(c["CertificateArn"] == cfg["originCertificateArn"]
                  for c in listeners[0].get("Certificates", [])))
    targets = cloud.client("elbv2").describe_target_health(
        TargetGroupArn=app["TargetGroupArn"])["TargetHealthDescriptions"]
    check("healthy-target", len(targets) == 1
          and all(t["TargetHealth"]["State"] == "healthy" for t in targets))
    # Sensitive origin headers remain in memory; output contains booleans only.
    distribution = cloud.client("cloudfront", "us-east-1").get_distribution(
        Id=app["DistributionId"])["Distribution"]
    settings = distribution["DistributionConfig"]
    check("cloudfront-deployed", distribution["Status"] == "Deployed" and settings["Enabled"])
    check("public-hostname", settings.get("Aliases", {}).get("Items") == [cfg["viewerDomainName"]]
          and settings["ViewerCertificate"].get("ACMCertificateArn") == cfg["viewerCertificateArn"])
    check("waf", settings.get("WebACLId") == cloud.outputs("edge")["WebAclArn"])
    origin = next(o for o in settings["Origins"]["Items"] if o["Id"] == "api")
    check("native-alb-origin", origin["DomainName"] == alb["DNSName"]
          and origin.get("CustomOriginConfig", {}).get("OriginProtocolPolicy") == "https-only")
    headers = {h["HeaderName"].lower(): h["HeaderValue"]
               for h in origin.get("CustomHeaders", {}).get("Items", [])}
    rules = cloud.client("elbv2").describe_rules(ListenerArn=listeners[0]["ListenerArn"])["Rules"]
    expected = []
    for rule in rules:
        if not any(a["Type"] == "forward" and a.get("TargetGroupArn") == app["TargetGroupArn"]
                   for a in rule["Actions"]):
            continue
        for condition in rule.get("Conditions", []):
            header = condition.get("HttpHeaderConfig", {})
            if header.get("HttpHeaderName", "").lower() == "x-origin-verify":
                expected.extend(header.get("Values", []))
    check("origin-gate", len(expected) == 1 and bool(headers.get("x-origin-verify"))
          and hmac.compare_digest(expected[0], headers["x-origin-verify"])
          and listeners[0]["DefaultActions"][0]["Type"] == "fixed-response"
          and listeners[0]["DefaultActions"][0]["FixedResponseConfig"]["StatusCode"] == "403")
    behavior = next(b for b in settings["CacheBehaviors"]["Items"] if b["PathPattern"] == "/api/*")
    policy = cloud.client("cloudfront", "us-east-1").get_origin_request_policy(
        Id=behavior["OriginRequestPolicyId"])["OriginRequestPolicy"]["OriginRequestPolicyConfig"]
    forwarded = {h.lower() for h in policy["HeadersConfig"].get("Headers", {}).get("Items", [])}
    check("host-cookie-csrf-forwarded",
          {"host", "origin", "content-type", "x-csrf-token"}.issubset(forwarded)
          and policy["CookiesConfig"]["CookieBehavior"] == "all")
    for path in ("/api/*", "/healthz", "/readyz"):
        b = next(b for b in settings["CacheBehaviors"]["Items"] if b["PathPattern"] == path)
        cache = cloud.client("cloudfront", "us-east-1").get_cache_policy(
            Id=b["CachePolicyId"])["CachePolicy"]["CachePolicyConfig"]
        check("uncached:" + path, all(cache[k] == 0 for k in ("MinTTL", "DefaultTTL", "MaxTTL")))
    static = next(o for o in settings["Origins"]["Items"] if o["Id"] != "api")
    check("static-oac", bool(static.get("OriginAccessControlId"))
          and app["AssetsBucketName"] in static["DomainName"])
    for bucket in (foundation["ReplayBucketName"], app["AssetsBucketName"]):
        block = cloud.client("s3").get_public_access_block(
            Bucket=bucket)["PublicAccessBlockConfiguration"]
        encryption = cloud.client("s3").get_bucket_encryption(
            Bucket=bucket)["ServerSideEncryptionConfiguration"]["Rules"]
        check("private-encrypted-bucket:" + bucket, all(block.values())
              and any(r["ApplyServerSideEncryptionByDefault"]["SSEAlgorithm"] in
                      ("AES256", "aws:kms") for r in encryption))
    ttl = cloud.client("dynamodb").describe_time_to_live(
        TableName=foundation["RunsTableName"])["TimeToLiveDescription"]
    check("run-ttl", ttl.get("AttributeName") == "expiresAt"
          and ttl.get("TimeToLiveStatus") in ("ENABLED", "ENABLING"))
    log_prefixes = [(app["LogGroupName"], True)]
    if cfg["plannerMode"] == "agentcore":
        agent = cloud.outputs("agentcore")
        runtime = cloud.client("bedrock-agentcore-control").get_agent_runtime(
            agentRuntimeId=agent["AgentRuntimeId"])
        check("runtime-ready", runtime["status"] == "READY")
        log_prefixes.append(("/aws/bedrock-agentcore/runtimes/" + agent["AgentRuntimeId"], False))
    for prefix, exact in log_prefixes:
        logs = []
        for page in cloud.client("logs").get_paginator("describe_log_groups").paginate(logGroupNamePrefix=prefix):
            logs.extend(g for g in page["logGroups"] if not exact or g["logGroupName"] == prefix)
        check("log-retention:" + prefix, logs and all(g.get("retentionInDays") == cfg["logRetentionDays"]
                                                     for g in logs))
    if not args.skip_http:
        base = app["PublicUrl"]
        check("owned-public-url", base == "https://" + cfg["viewerDomainName"])
        opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(CookieJar()))
        for path in ("/", "/healthz", "/readyz", "/api/config", "/api/runs"):
            with opener.open(base + path, timeout=30) as response:
                body = response.read(2 * 1024 * 1024)
                passed = response.status == 200
                if path.startswith("/api/"):
                    passed = passed and "no-store" in response.headers.get("Cache-Control", "")
                if path == "/api/config":
                    value = json.loads(body)
                    cookie = next((c for c in response.headers.get_all("Set-Cookie", [])
                                   if c.startswith(PROJECT.replace("-", "_") + "_session=")), "")
                    passed = passed and value["planner"] == cfg["plannerMode"]
                    passed = passed and value["storage"] == "aws" and value["simulation"] == "ready"
                    passed = passed and all(flag in cookie.lower()
                                             for flag in ("secure", "httponly", "samesite=strict"))
                check("https:" + path, passed)
    print(json.dumps({"verification": "passed", "checks": len(checks),
                      "modelInvocation": False, "runCreation": False,
                      "httpChecked": not args.skip_http}))

if __name__ == "__main__":
    try:
        main()
    except Exception as error:
        print(json.dumps({"verification": "failed", "error": type(error).__name__,
                          "next": "Inspect the last check and owned resource status"}))
        sys.exit(1)
PY
)

cd "$ROBOT_LAB"
test -f infra/workshop-binding.json
(
set -o noclobber
cat > .local/tools/verify-demo.py <<'PY'
"""Two live planner calls and one physical run, scoped to this participant."""
import argparse
import base64
from http.cookiejar import CookieJar
import json
from pathlib import Path
import sys
import time
import urllib.error
import urllib.request

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--seed", type=int, default=123)
    args = parser.parse_args()
    if not 0 <= args.seed <= 9999:
        raise ValueError("Seed must be 0..9999")
    root = Path.cwd()
    if not (root / "infra/workshop-binding.json").is_file():
        raise ValueError("Run from the prepared participant app")
    sys.path.insert(0, str(root / "infra"))
    from deployment import Cloud, PROJECT, read_config
    cloud = Cloud(read_config(root / "infra/config.json"))
    base = cloud.outputs("app")["PublicUrl"]
    assert base == "https://" + cloud.config["viewerDomainName"]
    cookies, checks = CookieJar(), []
    opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cookies))
    csrf = ""
    def request(path, payload=None, client=opener):
        headers = {"Origin": base, "Accept": "application/json"}
        data = None
        if payload is not None:
            headers.update({"Content-Type": "application/json", "X-CSRF-Token": csrf})
            data = json.dumps(payload, ensure_ascii=False).encode()
        req = urllib.request.Request(base + path, data=data, headers=headers)
        with client.open(req, timeout=120) as response:
            assert response.headers.get_content_type() == "application/json"
            body = response.read(16 * 1024 * 1024 + 1)
            assert len(body) <= 16 * 1024 * 1024
            return json.loads(body)
    def check(name, passed):
        checks.append({"check": name, "passed": bool(passed)})
        print(json.dumps(checks[-1]), flush=True)
        if not passed:
            raise AssertionError(name)
    config = request("/api/config")
    csrf = config["csrf"]
    check("live-config", config["storage"] == "aws" and config["simulation"] == "ready"
          and config["planner"] == cloud.config["plannerMode"] == "agentcore")
    latencies = []
    for _ in range(2):
        start = time.perf_counter()
        plan = request("/api/plan", {
            "prompt": "파란 상자, 빨간 상자, 주황 상자 순서로 3단으로 쌓아줘",
            "scenario": "stack"})
        latencies.append(round(time.perf_counter() - start, 3))
        check("plan-order", plan["scenario"] == "stack"
              and [a["objectId"] for a in plan["actions"]] == ["blue", "red", "orange"]
              and all(a["target"] == "tower" for a in plan["actions"]))
        check("plan-provider", plan["provider"] == "agentcore")
    start = time.perf_counter()
    run = request("/api/runs", {"planId": plan["id"], "seed": args.seed})
    run_id = run["id"]
    deadline = time.monotonic() + 180
    while run["status"] in ("queued", "running") and time.monotonic() < deadline:
        time.sleep(1)
        run = request("/api/runs/" + run_id)
    elapsed = round(time.perf_counter() - start, 3)
    check("run-succeeded", run["status"] == "succeeded")
    replay = request("/api/runs/" + run_id + "/replay")
    check("fresh-mujoco", replay["engine"] == "MuJoCo" and replay["seed"] == args.seed)
    check("physical-checks", replay["status"] == "succeeded"
          and replay["metrics"]["success"] and replay["metrics"]["placed"] == 3
          and replay["checks"] and all(c["passed"] for c in replay["checks"]))
    bodies = replay["frames"][-1]["bodies"]
    heights = [bodies[name]["position"][2] for name in ("blue", "red", "orange")]
    check("released-tower-order", heights[0] < heights[1] < heights[2]
          and all(abs(heights[i + 1] - heights[i] - 0.05) < 0.008 for i in range(2)))
    token = next(c.value for c in cookies if c.name == PROJECT.replace("-", "_") + "_session")
    encoded = token.split(".")[1]
    owner = json.loads(base64.urlsafe_b64decode(encoded + "=" * (-len(encoded) % 4)))["id"]
    foundation = cloud.outputs("foundation")
    item = cloud.client("dynamodb").get_item(
        TableName=foundation["RunsTableName"], ConsistentRead=True,
        Key={"pk": {"S": "SESSION#" + owner}, "sk": {"S": "RUN#" + run_id}}).get("Item", {})
    check("dynamodb-persisted", item.get("run", {}).get("M", {}).get("status", {}).get("S") == "succeeded")
    stored = cloud.client("s3").get_object(
        Bucket=foundation["ReplayBucketName"], Key=f"sessions/{owner}/{run_id}.json")
    check("s3-api-match", json.load(stored["Body"]) == replay)
    guest = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(CookieJar()))
    request("/api/config", client=guest)
    try:
        request("/api/runs/" + run_id + "/replay", client=guest)
        isolated = False
    except urllib.error.HTTPError as error:
        isolated = error.code == 404
    check("other-session-404", isolated)
    report = {
        "passed": True, "publicUrl": base, "planner": config["planner"],
        "runId": run_id, "seed": args.seed, "metrics": replay["metrics"], "checks": checks,
        "planRequestSeconds": latencies, "executionSeconds": elapsed,
        "latencyScope": "Includes model inference and network; not isolated Runtime overhead",
    }
    Path(".local/evidence").mkdir(parents=True, exist_ok=True)
    Path(".local/evidence/live-verification.json").write_text(
        json.dumps(report, ensure_ascii=False, indent=2) + "\n")
    print(json.dumps(report, ensure_ascii=False))

if __name__ == "__main__":
    try:
        main()
    except Exception as error:
        print(json.dumps({"passed": False, "error": type(error).__name__,
                          "httpStatus": getattr(error, "code", None),
                          "next": "Inspect the last check; no credentials or session values emitted"}))
        sys.exit(1)
PY
)

python3 -m py_compile .local/tools/verify-aws.py .local/tools/verify-demo.py
python3 .local/tools/verify-aws.py --help
python3 .local/tools/verify-demo.py --help